MB-34017: Flush HCS from Disk Checkpoint

BenHuddleston · daverigby · commit a2b77488bc74 · 2019-08-19T09:08:13.000Z
Flush the HCS stored in Disk checkpoints when we flush Disk checkpoints so that we have a correct HCS which will be used to optimise warmup as often as possible. Change-Id: I6db95f60f82259ebbedcae048824ca2d885a8e93 Reviewed-on: http://review.couchbase.org/113255 Reviewed-by: James Harrison <james.harrison@couchbase.com> Reviewed-by: Dave Rigby <daver@couchbase.com> Tested-by: Build Bot <build@couchbase.com>
diff --git a/engines/ep/src/checkpoint.cc b/engines/ep/src/checkpoint.cc
@@ -94,6 +94,7 @@ Checkpoint::Checkpoint(EPStats& st,
                        uint64_t id,
                        uint64_t snapStart,
                        uint64_t snapEnd,
+                       boost::optional<uint64_t> highCompletedSeqno,
                        Vbid vbid,
                        CheckpointType checkpointType)
     : stats(st),
@@ -110,7 +111,8 @@ Checkpoint::Checkpoint(EPStats& st,
       metaKeyIndex(keyIndexTrackingAllocator),
       keyIndexMemUsage(0),
       queuedItemsMemUsage(0),
-      checkpointType(checkpointType) {
+      checkpointType(checkpointType),
+      highCompletedSeqno(highCompletedSeqno) {
     stats.coreLocal.get()->memOverhead.fetch_add(sizeof(Checkpoint));
 }
 
diff --git a/engines/ep/src/checkpoint.h b/engines/ep/src/checkpoint.h
@@ -23,6 +23,7 @@
 #include "item.h"
 #include "monotonic.h"
 
+#include <boost/optional.hpp>
 #include <folly/Synchronized.h>
 #include <platform/non_negative_counter.h>
 #include <utilities/memory_tracking_allocator.h>
@@ -394,6 +395,7 @@ class Checkpoint {
                uint64_t id,
                uint64_t snapStart,
                uint64_t snapEnd,
+               boost::optional<uint64_t> highCompletedSeqno,
                Vbid vbid,
                CheckpointType checkpointType);
 
@@ -519,6 +521,14 @@ class Checkpoint {
         checkpointType = type;
     }
 
+    void setHighCompletedSeqno(boost::optional<uint64_t> seqno) {
+        highCompletedSeqno = seqno;
+    }
+
+    boost::optional<uint64_t> getHighCompletedSeqno() {
+        return highCompletedSeqno;
+    }
+
     /**
      * Returns an iterator pointing to the beginning of the CheckpointQueue,
      * toWrite.
@@ -639,5 +649,12 @@ class Checkpoint {
     // Is this a checkpoint created by a replica from a received disk snapshot?
     CheckpointType checkpointType;
 
+    // The SyncRep HCS for this checkpoint. Used to ensure that we flush a
+    // correct HCS at the end of a snapshot to disk. This is optional as it is
+    // only necessary for Disk snapshot (due to de-dupe) and the way we retrieve
+    // items from the CheckpointManager for memory snapshots makes it
+    // non-trivial to send the HCS in memory snapshot markers.
+    boost::optional<uint64_t> highCompletedSeqno;
+
     friend std::ostream& operator <<(std::ostream& os, const Checkpoint& m);
 };
diff --git a/engines/ep/src/checkpoint_manager.cc b/engines/ep/src/checkpoint_manager.cc
@@ -52,7 +52,8 @@ CheckpointManager::CheckpointManager(EPStats& st,
     //     when the checkpointList is empty.
     //     Only in CheckpointManager::clear_UNLOCKED, the checkpointList
     //     is temporarily cleared and a new open checkpoint added immediately.
-    addOpenCheckpoint(1, lastSnapStart, lastSnapEnd, CheckpointType::Memory);
+    addOpenCheckpoint(
+            1, lastSnapStart, lastSnapEnd, {}, CheckpointType::Memory);
 
     if (checkpointConfig.isPersistenceEnabled()) {
         // Register the persistence cursor
@@ -137,13 +138,14 @@ Checkpoint& CheckpointManager::getOpenCheckpoint_UNLOCKED(
 
 void CheckpointManager::addNewCheckpoint_UNLOCKED(uint64_t id) {
     addNewCheckpoint_UNLOCKED(
-            id, lastBySeqno, lastBySeqno, CheckpointType::Memory);
+            id, lastBySeqno, lastBySeqno, {}, CheckpointType::Memory);
 }
 
 void CheckpointManager::addNewCheckpoint_UNLOCKED(
         uint64_t id,
         uint64_t snapStartSeqno,
         uint64_t snapEndSeqno,
+        boost::optional<uint64_t> highCompletedSeqno,
         CheckpointType checkpointType) {
     // First, we must close the open checkpoint.
     auto& oldOpenCkpt = *checkpointList.back();
@@ -169,7 +171,11 @@ void CheckpointManager::addNewCheckpoint_UNLOCKED(
             id,
             snapStartSeqno,
             snapEndSeqno);
-    addOpenCheckpoint(id, snapStartSeqno, snapEndSeqno, checkpointType);
+    addOpenCheckpoint(id,
+                      snapStartSeqno,
+                      snapEndSeqno,
+                      highCompletedSeqno,
+                      checkpointType);
 
     /* If cursors reached to the end of its current checkpoint, move it to the
        next checkpoint. DCP and Persistence cursors can skip a "checkpoint end"
@@ -202,16 +208,23 @@ void CheckpointManager::addNewCheckpoint_UNLOCKED(
     }
 }
 
-void CheckpointManager::addOpenCheckpoint(uint64_t id,
-                                          uint64_t snapStart,
-                                          uint64_t snapEnd,
-                                          CheckpointType checkpointType) {
+void CheckpointManager::addOpenCheckpoint(
+        uint64_t id,
+        uint64_t snapStart,
+        uint64_t snapEnd,
+        boost::optional<uint64_t> highCompletedSeqno,
+        CheckpointType checkpointType) {
     Expects(checkpointList.empty() ||
             checkpointList.back()->getState() ==
                     checkpoint_state::CHECKPOINT_CLOSED);
 
-    auto ckpt = std::make_unique<Checkpoint>(
-            stats, id, snapStart, snapEnd, vbucketId, checkpointType);
+    auto ckpt = std::make_unique<Checkpoint>(stats,
+                                             id,
+                                             snapStart,
+                                             snapEnd,
+                                             highCompletedSeqno,
+                                             vbucketId,
+                                             checkpointType);
     // 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
@@ -842,6 +855,7 @@ CheckpointManager::ItemsForCursor CheckpointManager::getItemsForCursor(
     // limit.
     ItemsForCursor result((*cursor.currentCheckpoint)->getSnapshotStartSeqno(),
                           (*cursor.currentCheckpoint)->getSnapshotEndSeqno(),
+                          (*cursor.currentCheckpoint)->getHighCompletedSeqno(),
                           (*cursor.currentCheckpoint)->getCheckpointType());
 
     size_t itemCount = 0;
@@ -868,6 +882,11 @@ CheckpointManager::ItemsForCursor CheckpointManager::getItemsForCursor(
                 result.range.setEnd(
                         (*cursor.currentCheckpoint)->getSnapshotEndSeqno());
 
+                // Only move the HCS at checkpoint end (don't want to flush a
+                // HCS mid-checkpoint).
+                result.highCompletedSeqno =
+                        (*cursor.currentCheckpoint)->getHighCompletedSeqno();
+
                 // However, we *do* want to move the cursor into the next
                 // checkpoint if possible; as that means the checkpoint we just
                 // completed has one less cursor in it (and could potentially be
@@ -952,6 +971,7 @@ void CheckpointManager::clear_UNLOCKED(vbucket_state_t vbState, uint64_t seqno)
     addOpenCheckpoint(vbucket_state_active ? 1 : 0 /* id */,
                       lastBySeqno,
                       lastBySeqno,
+                      {},
                       CheckpointType::Memory);
     resetCursors();
 }
@@ -1077,9 +1097,11 @@ void CheckpointManager::setBackfillPhase(uint64_t start, uint64_t end) {
     openCkpt.setSnapshotEndSeqno(end);
 }
 
-void CheckpointManager::createSnapshot(uint64_t snapStartSeqno,
-                                       uint64_t snapEndSeqno,
-                                       CheckpointType checkpointType) {
+void CheckpointManager::createSnapshot(
+        uint64_t snapStartSeqno,
+        uint64_t snapEndSeqno,
+        boost::optional<uint64_t> highCompletedSeqno,
+        CheckpointType checkpointType) {
     LockHolder lh(queueLock);
 
     auto& openCkpt = getOpenCheckpoint_UNLOCKED(lh);
@@ -1093,11 +1115,15 @@ void CheckpointManager::createSnapshot(uint64_t snapStartSeqno,
         openCkpt.setSnapshotStartSeqno(snapStartSeqno);
         openCkpt.setSnapshotEndSeqno(snapEndSeqno);
         openCkpt.setCheckpointType(checkpointType);
+        openCkpt.setHighCompletedSeqno(highCompletedSeqno);
         return;
     }
 
-    addNewCheckpoint_UNLOCKED(
-            openCkptId + 1, snapStartSeqno, snapEndSeqno, checkpointType);
+    addNewCheckpoint_UNLOCKED(openCkptId + 1,
+                              snapStartSeqno,
+                              snapEndSeqno,
+                              highCompletedSeqno,
+                              checkpointType);
 }
 
 void CheckpointManager::resetSnapshotRange() {
diff --git a/engines/ep/src/checkpoint_manager.h b/engines/ep/src/checkpoint_manager.h
@@ -23,6 +23,7 @@
 #include "monotonic.h"
 #include "queue_op.h"
 
+#include <boost/optional.hpp>
 #include <memcached/engine_common.h>
 #include <memcached/vbucket.h>
 #include <memory>
@@ -58,11 +59,17 @@ class CheckpointManager {
     struct ItemsForCursor {
         ItemsForCursor(uint64_t start,
                        uint64_t end,
+                       boost::optional<uint64_t> highCompletedSeqno = {},
                        CheckpointType checkpointType = CheckpointType::Memory)
-            : range(start, end), checkpointType(checkpointType) {
+            : range(start, end),
+              highCompletedSeqno(highCompletedSeqno),
+              checkpointType(checkpointType) {
         }
         snapshot_range_t range;
         bool moreAvailable = {false};
+
+        // HCS that should be flushed
+        boost::optional<uint64_t> highCompletedSeqno = {};
         CheckpointType checkpointType = CheckpointType::Memory;
     };
 
@@ -338,6 +345,7 @@ class CheckpointManager {
 
     void createSnapshot(uint64_t snapStartSeqno,
                         uint64_t snapEndSeqno,
+                        boost::optional<uint64_t> highCompletedSeqno,
                         CheckpointType checkpointType);
 
     void resetSnapshotRange();
@@ -421,12 +429,14 @@ class CheckpointManager {
      * @param id for the new checkpoint
      * @param snapStartSeqno for the new checkpoint
      * @param snapEndSeqno for the new checkpoint
+     * @param highCompletedSeqno optional SyncRep HCS to be flushed to disk
      * @param checkpointType is the checkpoint created from a replica receiving
      *                       a disk snapshot?
      */
     void addNewCheckpoint_UNLOCKED(uint64_t id,
                                    uint64_t snapStartSeqno,
                                    uint64_t snapEndSeqno,
+                                   boost::optional<uint64_t> highCompletedSeqno,
                                    CheckpointType checkpointType);
 
     /*
@@ -445,12 +455,14 @@ class CheckpointManager {
      * @param id for the new checkpoint
      * @param snapStartSeqno for the new checkpoint
      * @param snapEndSeqno for the new checkpoint
+     * @param highCompletedSeqno the SyncRepl HCS to be flushed to disk
      * @param checkpointType is the checkpoint created from a replica receiving
      *                       a disk snapshot?
      */
     void addOpenCheckpoint(uint64_t id,
                            uint64_t snapStart,
                            uint64_t snapEnd,
+                           boost::optional<uint64_t> highCompletedSeqno,
                            CheckpointType checkpointType);
 
     bool moveCursorToNextCheckpoint(CheckpointCursor &cursor);
diff --git a/engines/ep/src/dcp/passive_stream.cc b/engines/ep/src/dcp/passive_stream.cc
@@ -896,12 +896,14 @@ void PassiveStream::processMarker(SnapshotMarker* marker) {
             vb->setReceivingInitialDiskSnapshot(true);
             ckptMgr.createSnapshot(cur_snapshot_start.load(),
                                    cur_snapshot_end.load(),
+                                   marker->getHighCompletedSeqno(),
                                    checkpointType);
         } else {
             if (marker->getFlags() & MARKER_FLAG_CHK ||
                 vb->checkpointManager->getOpenCheckpointId() == 0) {
                 ckptMgr.createSnapshot(cur_snapshot_start.load(),
                                        cur_snapshot_end.load(),
+                                       marker->getHighCompletedSeqno(),
                                        checkpointType);
             } else {
                 // If we are reconnecting then we need to update the snap end
diff --git a/engines/ep/src/ep_bucket.cc b/engines/ep/src/ep_bucket.cc
@@ -409,8 +409,14 @@ std::pair<bool, size_t> EPBucket::flushVBucket(Vbid vbid) {
             Collections::VB::Flush collectionFlush(vb->getManifest());
 
             // HCS is optional because we have to update it on disk only if some
-            // Commit/Abort SyncWrite is found in the flush-batch
-            boost::optional<uint64_t> hcs;
+            // Commit/Abort SyncWrite is found in the flush-batch. If we're
+            // flushing Disk checkpoints then the toFlush value may be
+            // supplied. In this case, this should be the HCS received from the
+            // Active node and should be greater than or equal to the HCS for
+            // any other item in this flush batch. This is required because we
+            // send mutations instead of a commits and would not otherwise
+            // update the HCS on disk.
+            boost::optional<uint64_t> hcs = toFlush.highCompletedSeqno;
             // HPS is optional because we have to update it on disk only if a
             // prepare is found in the flush-batch
             boost::optional<uint64_t> hps;
diff --git a/engines/ep/src/vbucket.cc b/engines/ep/src/vbucket.cc
@@ -407,6 +407,7 @@ VBucket::ItemsToFlush VBucket::getItemsToPersist(size_t approxLimit) {
         auto ckptItems = checkpointManager->getItemsForPersistence(
                 result.items, ckptMgrLimit);
         result.range = ckptItems.range;
+        result.highCompletedSeqno = ckptItems.highCompletedSeqno;
         ckptItemsAvailable = ckptItems.moreAvailable;
         stats.persistenceCursorGetItemsHisto.add(
                 std::chrono::duration_cast<std::chrono::microseconds>(
diff --git a/engines/ep/src/vbucket.h b/engines/ep/src/vbucket.h
@@ -471,6 +471,7 @@ class VBucket : public std::enable_shared_from_this<VBucket> {
         std::vector<queued_item> items;
         snapshot_range_t range{0, 0};
         bool moreAvailable = false;
+        boost::optional<uint64_t> highCompletedSeqno = {};
     };
 
     /**
diff --git a/engines/ep/tests/module_tests/dcp_durability_stream_test.cc b/engines/ep/tests/module_tests/dcp_durability_stream_test.cc
@@ -968,6 +968,33 @@ TEST_P(DurabilityPassiveStreamPersistentTest,
     EXPECT_EQ(1, ack.getPreparedSeqno());
 }
 
+TEST_P(DurabilityPassiveStreamPersistentTest, DiskSnapshotHCSPersisted) {
+    testReceiveMutationOrDeletionInsteadOfCommitWhenStreamingFromDisk(
+            DocumentState::Alive);
+    flushVBucketToDiskIfPersistent(vbid, 2);
+    {
+        auto vb = store->getVBucket(vbid);
+        EXPECT_EQ(2, vb->getHighCompletedSeqno());
+    }
+
+    // Reset and warmup to check persistence
+    consumer->closeAllStreams();
+    consumer.reset();
+    resetEngineAndWarmup();
+    // Recreate the consumer so that our normal test TearDown will work
+    consumer =
+            std::make_shared<MockDcpConsumer>(*engine, cookie, "test_consumer");
+    consumer->enableSyncReplication();
+    consumer->addStream(0 /*opaque*/, vbid, 0 /*flags*/);
+    stream = static_cast<MockPassiveStream*>(
+            (consumer->getVbucketStream(vbid)).get());
+
+    {
+        auto vb = store->getVBucket(vbid);
+        EXPECT_EQ(2, vb->getHighCompletedSeqno());
+    }
+}
+
 TEST_P(DurabilityPassiveStreamTest,
        NoSeqnoAckOnStreamAcceptanceIfNotSupported) {
     consumer->disableSyncReplication();
@@ -1007,7 +1034,7 @@ void DurabilityPassiveStreamTest::
                           2 /*snapStart*/,
                           4 /*snapEnd*/,
                           dcp_marker_flag_t::MARKER_FLAG_DISK | MARKER_FLAG_CHK,
-                          {} /*HCS*/,
+                          2 /*HCS*/,
                           {} /*streamId*/);
     stream->processMarker(&marker);
 
diff --git a/engines/ep/tests/module_tests/durability_monitor_test.cc b/engines/ep/tests/module_tests/durability_monitor_test.cc
diff --git a/engines/ep/tests/module_tests/evp_store_durability_test.cc b/engines/ep/tests/module_tests/evp_store_durability_test.cc
diff --git a/engines/ep/tests/module_tests/evp_store_warmup_test.cc b/engines/ep/tests/module_tests/evp_store_warmup_test.cc
diff --git a/engines/ep/tests/module_tests/vbucket_durability_test.cc b/engines/ep/tests/module_tests/vbucket_durability_test.cc
