[BP] MB-37321: Register cursor before backfilling after cursor dropping

jameseh96 · jimwwalker · commit 92f1f76d6d4a · 2020-01-10T10:20:23.000Z
Prior to this patch there was a small window of time after ActiveStream checked the persisted seqno to determine if a backfill could provide any items needed to catch up to the checkpoint manager _before_ the cursor is reregistered. In this window, closed checkpoints might be removed, or checkpoint expelling might remove some items from the oldest checkpoint. Either of these events would lead to the stream "skipping" some items; the cursor would be registered at a later seqno than expected, and the stream does not re-check whether a backfill is needed. This patch makes ActiveStream register the cursor before trying to schedule a backfill. If the backfill end extends later than the requested seqno (an expected situation) existing code in markDiskSnapshot "atomically" (under the checkpoint manager lock) reregisters a cursor at the correct seqno. Change-Id: I472eaf97537532fabb0eed323a4789342a547a98 Reviewed-on: http://review.couchbase.org/120038 Tested-by: Build Bot <build@couchbase.com> Well-Formed: Build Bot <build@couchbase.com> Reviewed-by: Jim Walker <jim@couchbase.com>
diff --git a/engines/ep/src/checkpoint.h b/engines/ep/src/checkpoint.h
@@ -94,6 +94,7 @@ typedef std::list<std::pair<std::string, MustSendCheckpointEnd>>
 class Checkpoint;
 class CheckpointManager;
 class CheckpointConfig;
+class CheckpointCursorIntrospector;
 class PreLinkDocumentContext;
 class VBucket;
 
@@ -130,6 +131,8 @@ using CheckpointList = std::list<std::unique_ptr<Checkpoint>>;
 class CheckpointCursor {
     friend class CheckpointManager;
     friend class Checkpoint;
+    friend class CheckpointCursorIntrospector;
+
 public:
 
     CheckpointCursor() { }
@@ -1062,6 +1065,7 @@ class CheckpointManager {
     FlusherCallback          flusherCB;
 
     friend std::ostream& operator<<(std::ostream& os, const CheckpointManager& m);
+    friend class CheckpointCursorIntrospector;
 };
 
 // Outputs a textual description of the CheckpointManager.
diff --git a/engines/ep/src/dcp/producer.cc b/engines/ep/src/dcp/producer.cc
@@ -929,11 +929,15 @@ void DcpProducer::recordBackfillManagerBytesSent(size_t bytes) {
     backfillMgr->bytesSent(bytes);
 }
 
-void DcpProducer::scheduleBackfillManager(VBucket& vb,
+bool DcpProducer::scheduleBackfillManager(VBucket& vb,
                                           std::shared_ptr<ActiveStream> s,
                                           uint64_t start,
                                           uint64_t end) {
-    backfillMgr->schedule(vb, s, start, end);
+    if (start <= end) {
+        backfillMgr->schedule(vb, s, start, end);
+        return true;
+    }
+    return false;
 }
 
 void DcpProducer::addStats(ADD_STAT add_stat, const void *c) {
diff --git a/engines/ep/src/dcp/producer.h b/engines/ep/src/dcp/producer.h
@@ -143,10 +143,10 @@ class DcpProducer : public ConnHandler,
     void notifyBackfillManager();
     bool recordBackfillManagerBytesRead(size_t bytes, bool force);
     void recordBackfillManagerBytesSent(size_t bytes);
-    void scheduleBackfillManager(VBucket& vb,
-                                 std::shared_ptr<ActiveStream> s,
-                                 uint64_t start,
-                                 uint64_t end);
+    virtual bool scheduleBackfillManager(VBucket& vb,
+                                         std::shared_ptr<ActiveStream> s,
+                                         uint64_t start,
+                                         uint64_t end);
 
     bool isExtMetaDataEnabled () {
         return enableExtMetaData;
diff --git a/engines/ep/src/dcp/stream.cc b/engines/ep/src/dcp/stream.cc
@@ -1444,49 +1444,61 @@ void ActiveStream::scheduleBackfill_UNLOCKED(bool reschedule) {
     }
 
     uint64_t backfillStart = lastReadSeqno.load() + 1;
-    uint64_t backfillEnd = 0;
-    bool tryBackfill = false;
+    uint64_t backfillEnd;
+    bool tryBackfill;
 
-    if ((flags_ & DCP_ADD_STREAM_FLAG_DISKONLY) || reschedule) {
-        uint64_t vbHighSeqno = static_cast<uint64_t>(vbucket->getHighSeqno());
-        if (lastReadSeqno.load() > vbHighSeqno) {
-            throw std::logic_error("ActiveStream::scheduleBackfill_UNLOCKED: "
-                                   "lastReadSeqno (which is " +
-                                   std::to_string(lastReadSeqno.load()) +
-                                   " ) is greater than vbHighSeqno (which is " +
-                                   std::to_string(vbHighSeqno) + " ). " +
-                                   "for stream " + producer->logHeader() +
-                                   "; vb " + std::to_string(vb_));
-        }
-        if (reschedule) {
-            /* We need to do this for reschedule because in case of
-               DCP_ADD_STREAM_FLAG_DISKONLY (the else part), end_seqno_ is
-               set to last persisted seqno befor calling
-               scheduleBackfill_UNLOCKED() */
-            backfillEnd = engine->getKVBucket()->getLastPersistedSeqno(vb_);
-        } else {
-            backfillEnd = end_seqno_;
-        }
+    if (flags_ & static_cast<uint64_t>(DCP_ADD_STREAM_FLAG_DISKONLY)) {
+        // if disk only, always backfill to the requested end seqno
+        backfillEnd = end_seqno_;
         tryBackfill = true;
     } else {
+        /* not disk only - stream may require backfill but will transition to
+         * in-memory afterward; register the cursor now.
+         * There are two expected cases:
+         *  1: registerResult.second=true, which means
+         *     - Cursor at start of first checkpoint
+         *     - CheckpointManager can't provide all the items needed
+         *       so a backfill may be required before moving to
+         *       in-memory streaming.
+         *  2: registerResult.second=false
+         *     - The CheckpointManager contains the required items
+         *     - No backfill needed
+         */
+
+        CursorRegResult registerResult;
         try {
-            std::tie(curChkSeqno, tryBackfill) =
-                    vbucket->checkpointManager->registerCursorBySeqno(
-                            cursorName,
-                            lastReadSeqno.load(),
-                            MustSendCheckpointEnd::NO);
-            cursorRegistered = true;
-        } catch(std::exception& error) {
+            registerResult = vbucket->checkpointManager->registerCursorBySeqno(
+                    cursorName,
+                    lastReadSeqno.load(),
+                    MustSendCheckpointEnd::NO);
+        } catch (std::exception& error) {
             log(EXTENSION_LOG_WARNING,
                 "(vb %" PRIu16
                 ") Failed to register "
                 "cursor: %s",
                 vb_,
                 error.what());
             endStream(END_STREAM_STATE);
+            return;
         }
 
+        curChkSeqno = registerResult.first;
+        tryBackfill = registerResult.second;
+        cursorRegistered = true;
+
+        log(EXTENSION_LOG_INFO,
+            "(vb %" PRIu16
+            ") ActiveStream::scheduleBackfill_UNLOCKED register cursor "
+            "with "
+            "name \"{}\" backfill:{}, seqno:{}",
+            vb_,
+            name_.c_str(),
+            tryBackfill,
+            curChkSeqno.load());
+
         if (lastReadSeqno.load() > curChkSeqno) {
+            // something went wrong registering the cursor - it is too early
+            // and could read items this stream has already sent.
             throw std::logic_error("ActiveStream::scheduleBackfill_UNLOCKED: "
                                    "lastReadSeqno (which is " +
                                    std::to_string(lastReadSeqno.load()) +
@@ -1496,23 +1508,18 @@ void ActiveStream::scheduleBackfill_UNLOCKED(bool reschedule) {
                                    "; vb " + std::to_string(vb_));
         }
 
-        /* We need to find the minimum seqno that needs to be backfilled in
-         * order to make sure that we don't miss anything when transitioning
-         * to a memory snapshot. The backfill task will always make sure that
-         * the backfill end seqno is contained in the backfill.
-         */
-        if (backfillStart < curChkSeqno) {
-            if (curChkSeqno > end_seqno_) {
-                /* Backfill only is enough */
-                backfillEnd = end_seqno_;
-            } else {
-                /* Backfill + in-memory streaming */
-                backfillEnd = curChkSeqno - 1;
-            }
-        }
+        // _if_ a backfill is required, it should end either at the
+        // requested stream end seqno OR the seqno immediately
+        // before what the checkpoint manager can provide
+        // - whichever is lower.
+        backfillEnd = std::min(end_seqno_, curChkSeqno - 1);
     }
 
-    if (backfillStart <= backfillEnd && tryBackfill) {
+    if (tryBackfill &&
+        producer->scheduleBackfillManager(
+                *vbucket, shared_from_this(), backfillStart, backfillEnd)) {
+        // backfill will be needed to catch up to the items in the
+        // CheckpointManager
         log(EXTENSION_LOG_NOTICE,
             "(vb %" PRIu16
             ") Scheduling backfill "
@@ -1523,79 +1530,17 @@ void ActiveStream::scheduleBackfill_UNLOCKED(bool reschedule) {
             backfillStart,
             backfillEnd,
             reschedule ? "True" : "False");
-        producer->scheduleBackfillManager(
-                *vbucket, shared_from_this(), backfillStart, backfillEnd);
+
         isBackfillTaskRunning.store(true);
         /// Number of backfill items is unknown until the Backfill task
         /// completes the scan phase - reset backfillRemaining counter.
         backfillRemaining.reset();
     } else {
-        if (reschedule) {
-            // Infrequent code path, see comment below.
-            log(EXTENSION_LOG_NOTICE,
-                "(vb %" PRIu16
-                ") Did not schedule "
-                "backfill with reschedule : True, "
-                "tryBackfill : True; "
-                "backfillStart : %" PRIu64
-                ", "
-                "backfillEnd : %" PRIu64
-                ", "
-                "flags_ : %" PRIu32
-                ", "
-                "start_seqno_ : %" PRIu64
-                ", "
-                "end_seqno_ : %" PRIu64
-                ", "
-                "lastReadSeqno : %" PRIu64
-                ", "
-                "lastSentSeqno : %" PRIu64
-                ", "
-                "curChkSeqno : %" PRIu64
-                ", "
-                "itemsReady : %s",
-                vb_,
-                backfillStart,
-                backfillEnd,
-                flags_,
-                start_seqno_,
-                end_seqno_,
-                lastReadSeqno.load(),
-                lastSentSeqno.load(),
-                curChkSeqno.load(),
-                itemsReady ? "True" : "False");
-
-            /* Cursor was dropped, but we will not do backfill.
-             * This may happen in a corner case where, the memory usage is high
-             * due to other vbuckets and persistence cursor moves ahead of
-             * replication cursor to new checkpoint open but does not persist
-             * items yet.
-             *
-             * Because we dropped the cursor but did not do a backfill (and
-             * therefore did not re-register a cursor in markDiskSnapshot) we
-             * must re-register the cursor here.
-             */
-            try {
-                CursorRegResult result =
-                        vbucket->checkpointManager->registerCursorBySeqno(
-                                cursorName,
-                                lastReadSeqno.load(),
-                                MustSendCheckpointEnd::NO);
-                cursorRegistered = true;
-                curChkSeqno = result.first;
-            } catch (std::exception& error) {
-                log(EXTENSION_LOG_WARNING,
-                    "(vb %" PRIu16
-                    ") Failed to register "
-                    "cursor: %s",
-                    vb_,
-                    error.what());
-                endStream(END_STREAM_STATE);
-            }
-        }
-        if (flags_ & DCP_ADD_STREAM_FLAG_DISKONLY) {
+        // backfill not needed
+        if (flags_ & static_cast<uint64_t>(DCP_ADD_STREAM_FLAG_DISKONLY)) {
             endStream(END_STREAM_OK);
-        } else if (flags_ & DCP_ADD_STREAM_FLAG_TAKEOVER) {
+        } else if (flags_ &
+                   static_cast<uint64_t>(DCP_ADD_STREAM_FLAG_TAKEOVER)) {
             transitionState(StreamState::TakeoverSend);
         } else {
             transitionState(StreamState::InMemory);
diff --git a/engines/ep/src/dcp/stream.h b/engines/ep/src/dcp/stream.h
@@ -131,7 +131,6 @@ class Stream {
         clear_UNLOCKED();
     }
 
-protected:
 
     // The StreamState is protected as it needs to be accessed by sub-classes
     enum class StreamState {
@@ -148,6 +147,7 @@ class Stream {
 
     StreamState getState() const { return state_; }
 
+protected:
     void clear_UNLOCKED();
 
     /* To be called after getting streamMutex lock */
diff --git a/engines/ep/tests/mock/mock_dcp_producer.cc b/engines/ep/tests/mock/mock_dcp_producer.cc
@@ -27,7 +27,9 @@ std::shared_ptr<MockActiveStream> MockDcpProducer::mockActiveStreamRequest(
         uint64_t end_seqno,
         uint64_t vbucket_uuid,
         uint64_t snap_start_seqno,
-        uint64_t snap_end_seqno) {
+        uint64_t snap_end_seqno,
+        IncludeValue includeValue,
+        IncludeXattrs includeXattrs) {
     auto stream = std::make_shared<MockActiveStream>(
             static_cast<EventuallyPersistentEngine*>(&engine_),
             std::static_pointer_cast<MockDcpProducer>(shared_from_this()),
@@ -38,7 +40,9 @@ std::shared_ptr<MockActiveStream> MockDcpProducer::mockActiveStreamRequest(
             end_seqno,
             vbucket_uuid,
             snap_start_seqno,
-            snap_end_seqno);
+            snap_end_seqno,
+            includeValue,
+            includeXattrs);
     stream->setActive();
     if (!streams.insert(std::make_pair(vb.getId(), stream))) {
         throw std::logic_error(
@@ -47,4 +51,4 @@ std::shared_ptr<MockActiveStream> MockDcpProducer::mockActiveStreamRequest(
     }
     notifyStreamReady(vb.getId());
     return stream;
-}
+}
diff --git a/engines/ep/tests/mock/mock_dcp_producer.h b/engines/ep/tests/mock/mock_dcp_producer.h
@@ -161,5 +161,22 @@ class MockDcpProducer : public DcpProducer {
             uint64_t end_seqno,
             uint64_t vbucket_uuid,
             uint64_t snap_start_seqno,
-            uint64_t snap_end_seqno);
+            uint64_t snap_end_seqno,
+            IncludeValue includeValue = IncludeValue::Yes,
+            IncludeXattrs includeXattrs = IncludeXattrs::Yes);
+
+    bool scheduleBackfillManager(VBucket& vb,
+                                 std::shared_ptr<ActiveStream> s,
+                                 uint64_t start,
+                                 uint64_t end) override {
+        beforeScheduleBackfillCB(end);
+        return DcpProducer::scheduleBackfillManager(
+                vb, std::move(s), start, end);
+    }
+
+    void setBeforeScheduleBackfillCB(std::function<void(uint64_t)> backfillCB) {
+        beforeScheduleBackfillCB = std::move(backfillCB);
+    }
+
+    std::function<void(uint64_t)> beforeScheduleBackfillCB = [](uint64_t) {};
 };
diff --git a/engines/ep/tests/module_tests/checkpoint_utils.h b/engines/ep/tests/module_tests/checkpoint_utils.h
@@ -0,0 +1,32 @@
+/* -*- Mode: C++; tab-width: 4; c-basic-offset: 4; indent-tabs-mode: nil -*- */
+/*
+ *     Copyright 2018 Couchbase, Inc
+ *
+ *   Licensed under the Apache License, Version 2.0 (the "License");
+ *   you may not use this file except in compliance with the License.
+ *   You may obtain a copy of the License at
+ *
+ *       http://www.apache.org/licenses/LICENSE-2.0
+ *
+ *   Unless required by applicable law or agreed to in writing, software
+ *   distributed under the License is distributed on an "AS IS" BASIS,
+ *   WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ *   See the License for the specific language governing permissions and
+ *   limitations under the License.
+ */
+#pragma once
+
+#include "checkpoint.h"
+
+/**
+ * Stateless class used to gain privileged access into CheckpointCursor for
+ * testing purposes. This is a friend class of CheckpointCursor.
+ */
+class CheckpointCursorIntrospector {
+public:
+    static const CheckpointQueue::iterator& getCurrentPos(
+            const CheckpointManager& cm, const std::string& cursorName) {
+        auto& cursor = cm.connCursors.find(cursorName)->second;
+        return cursor.currentPos;
+    }
+};
diff --git a/engines/ep/tests/module_tests/dcp_test.cc b/engines/ep/tests/module_tests/dcp_test.cc
diff --git a/engines/ep/tests/module_tests/evp_store_single_threaded_test.cc b/engines/ep/tests/module_tests/evp_store_single_threaded_test.cc
diff --git a/engines/ep/tests/module_tests/kv_bucket_test.cc b/engines/ep/tests/module_tests/kv_bucket_test.cc
diff --git a/engines/ep/tests/module_tests/kv_bucket_test.h b/engines/ep/tests/module_tests/kv_bucket_test.h
