[BP] MB-25798: Re-register a dropped cursor if we don't backfill

owend74 · daverigby · commit 2513928d34d1 · 2017-09-21T08:47:36.000Z
After dropping a cursor, when scheduleBackfill_UNLOCKED is called but the backfill task does not need to be scheduled, it means the cursor is not re-registered in markDiskSnapshot. Therefore the cursor must therefore be re-registered from within scheduleBackfill_UNLOCKED. Change-Id: I4a643aed902316e0753555564b8bfd9b56291efe Reviewed-on: http://review.couchbase.org/83531 Well-Formed: Build Bot <build@couchbase.com> Tested-by: Build Bot <build@couchbase.com> Reviewed-by: Dave Rigby <daver@couchbase.com>
diff --git a/src/checkpoint.cc b/src/checkpoint.cc
@@ -678,9 +678,9 @@ CursorRegResult CheckpointManager::registerCursorBySeqno(
          * number we are looking for is higher than anything currently assigned
          *  and there is already an assert above for this case.
          */
-        LOG(EXTENSION_LOG_WARNING, "Cursor not registered into vb %d "
-            " for stream '%s' because seqno %" PRIu64 " is too high",
-            vbucketId, name.c_str(), startBySeqno);
+        throw std::logic_error(
+                "CheckpointManager::registerCursorBySeqno the sequences number "
+                "is higher than anything currently assigned");
     }
     return result;
 }
diff --git a/src/dcp/stream.cc b/src/dcp/stream.cc
@@ -62,6 +62,14 @@ Stream::~Stream() {
     clear_UNLOCKED();
 }
 
+bool Stream::isBackfilling() const {
+    return state_.load() == STREAM_BACKFILLING;
+}
+
+bool Stream::isInMemory() const {
+    return state_.load() == STREAM_IN_MEMORY;
+}
+
 void Stream::clear_UNLOCKED() {
     while (!readyQ.empty()) {
         DcpResponse* resp = readyQ.front();
@@ -325,12 +333,21 @@ void ActiveStream::markDiskSnapshot(uint64_t startSeqno, uint64_t endSeqno) {
     lastSentSnapEndSeqno.store(endSeqno, std::memory_order_relaxed);
 
     if (!(flags_ & DCP_ADD_STREAM_FLAG_DISKONLY)) {
-        // Only re-register the cursor if we still need to get memory snapshots
-        CursorRegResult result =
-            vb->checkpointManager.registerCursorBySeqno(
-                                                name_, chkCursorSeqno,
-                                                MustSendCheckpointEnd::NO);
-        curChkSeqno = result.first;
+        // Only re-register the cursor if we still need to get memory
+        // snapshots
+        try {
+            CursorRegResult result =
+                    vb->checkpointManager.registerCursorBySeqno(
+                            name_, chkCursorSeqno,
+                            MustSendCheckpointEnd::NO);
+
+            curChkSeqno = result.first;
+        } catch(std::exception& error) {
+            producer->getLogger().log(EXTENSION_LOG_WARNING,
+                    "(vb %" PRIu16 ") Failed to register cursor: %s",
+                    vb_, error.what());
+            endStream(END_STREAM_STATE);
+        }
     }
 
     lh.unlock();
@@ -1024,13 +1041,17 @@ void ActiveStream::scheduleBackfill_UNLOCKED(bool reschedule) {
         }
         tryBackfill = true;
     } else {
-        CursorRegResult result =
-            vbucket->checkpointManager.registerCursorBySeqno(
-                                                name_,
-                                                lastReadSeqno.load(),
-                                                MustSendCheckpointEnd::NO);
-        curChkSeqno = result.first;
-        tryBackfill = result.second;
+        try {
+            std::tie(curChkSeqno, tryBackfill) =
+                    vbucket->checkpointManager.registerCursorBySeqno(
+                            name_, lastReadSeqno.load(),
+                            MustSendCheckpointEnd::NO);
+        } catch(std::exception& error) {
+            producer->getLogger().log(EXTENSION_LOG_WARNING,
+                                      "(vb %" PRIu16 ") Failed to register "
+                                      "cursor: %s", vb_, error.what());
+            endStream(END_STREAM_STATE);
+        }
 
         if (lastReadSeqno.load() > curChkSeqno) {
             throw std::logic_error("ActiveStream::scheduleBackfill_UNLOCKED: "
@@ -1088,6 +1109,30 @@ void ActiveStream::scheduleBackfill_UNLOCKED(bool reschedule) {
                                       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(
+                            name_, lastReadSeqno.load(),
+                            MustSendCheckpointEnd::NO);
+
+                    curChkSeqno = result.first;
+            } catch (std::exception& error) {
+                producer->getLogger().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) {
             endStream(END_STREAM_OK);
@@ -1097,17 +1142,15 @@ void ActiveStream::scheduleBackfill_UNLOCKED(bool reschedule) {
             transitionState(STREAM_IN_MEMORY);
         }
         if (reschedule) {
-            /* 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.
-               Note: (1) We must not notify when we schedule backfill for the
-                         first time because the stream is not yet in producer
-                         conn list of streams 
-                     (2) It is not absolutely necessary to notify immediately
-                         as conn manager or an incoming items will cause a
-                         notification eventually, but wouldn't hurt to do so */
+            /*
+             * It is not absolutely necessary to notify immediately as conn
+             * manager or an incoming item will cause a notification eventually,
+             * but wouldn't hurt to do so.
+             *
+             * Note: must not notify when we schedule a backfill for the first
+             * time (i.e. when reschedule is false) because the stream is not
+             * yet in producer conn list of streams.
+             */
             bool inverse = false;
             if (itemsReady.compare_exchange_strong(inverse, true)) {
                 producer->notifyStreamReady(vb_);
diff --git a/src/dcp/stream.h b/src/dcp/stream.h
@@ -133,6 +133,12 @@ class Stream : public RCValue {
         return state_ != STREAM_DEAD;
     }
 
+    /// @Returns true if state_ is Backfilling
+    bool isBackfilling() const;
+
+    /// @Returns true if state_ is InMemory
+    bool isInMemory() const;
+
     void clear() {
         LockHolder lh(streamMutex);
         clear_UNLOCKED();
diff --git a/tests/mock/mock_stream.h b/tests/mock/mock_stream.h
@@ -86,4 +86,8 @@ class MockActiveStream : public ActiveStream {
     bool public_getPendingBackfill() const {
         return pendingBackfill;
     }
+
+    void transitionStateToBackfilling() {
+        transitionState(STREAM_BACKFILLING);
+    }
 };
diff --git a/tests/module_tests/evp_store_single_threaded_test.cc b/tests/module_tests/evp_store_single_threaded_test.cc
@@ -119,6 +119,70 @@ class SingleThreadedEPStoreTest : public EventuallyPersistentStoreTest {
     SingleThreadedExecutorPool* task_executor;
 };
 
+/**
+ * The following test checks to see if a cursor is re-registered after it is
+ * dropped in handleSlowStream. In particular the test is for when
+ * scheduleBackfill_UNLOCKED is called however the backfill task does not need
+ * to be scheduled and therefore the cursor is not re-registered in
+ * markDiskSnapshot.  The cursor must therefore be registered from within
+ * scheduleBackfill_UNLOCKED.
+ *
+ * At the end of the test we should have 2 cursors: 1 persistence cursor and 1
+ * DCP stream cursor.
+ */
+TEST_F(SingleThreadedEPStoreTest, MB22421_reregister_cursor) {
+    // Make vbucket active.
+    setVBucketStateAndRunPersistTask(vbid, vbucket_state_active);
+    auto vb = store->getVBuckets().getBucket(vbid);
+    auto& ckpt_mgr = vb->checkpointManager;
+
+    // Create a Mock Dcp producer
+    dcp_producer_t producer = new MockDcpProducer(*engine,
+                                                  cookie,
+                                                  "test_producer",
+                                                  /*notifyOnly*/false);
+    // Create a Mock Active Stream
+    stream_t stream = new MockActiveStream(
+                                           static_cast<EventuallyPersistentEngine*>(engine.get()),
+                                           producer,
+                                           producer->getName(),
+                                           /*flags*/0,
+                                           /*opaque*/0, vbid,
+                                           /*st_seqno*/0,
+                                           /*en_seqno*/~0,
+                                           /*vb_uuid*/0xabcd,
+                                           /*snap_start_seqno*/0,
+                                           /*snap_end_seqno*/~0);
+
+    MockActiveStream* mock_stream =
+    static_cast<MockActiveStream*>(stream.get());
+
+    mock_stream->transitionStateToBackfilling();
+    EXPECT_TRUE(mock_stream->isInMemory())
+    << "stream state should have transitioned to StreamInMemory";
+    // Have a persistence cursor and DCP cursor
+    EXPECT_EQ(2, ckpt_mgr.getNumOfCursors());
+
+    mock_stream->public_setBackfillTaskRunning(true);
+    mock_stream->transitionStateToBackfilling();
+    EXPECT_TRUE(mock_stream->isBackfilling())
+    << "stream state should not have transitioned to StreamBackfilling";
+    mock_stream->handleSlowStream();
+    // The call to handleSlowStream should result in setting pendingBackfill
+    // flag to true and the DCP cursor being dropped
+    EXPECT_TRUE(mock_stream->public_getPendingBackfill());
+    EXPECT_EQ(1, ckpt_mgr.getNumOfCursors());
+
+    mock_stream->public_setBackfillTaskRunning(false);
+
+    //schedule a backfill
+    mock_stream->next();
+    // Calling scheduleBackfill_UNLOCKED(reschedule == true) will not actually
+    // schedule a backfill task because backfillStart (is lastReadSeqno + 1) is
+    // 1 and backfillEnd is 0, however the cursor still needs to be
+    // re-registered.
+    EXPECT_EQ(2, ckpt_mgr.getNumOfCursors());
+}
 
 /**
  * Regression test for MB-22451: When handleSlowStream is called and in

Original file line number	Diff line number	Diff line change
`@@ -678,9 +678,9 @@ CursorRegResult CheckpointManager::registerCursorBySeqno(`
`678`	`678`	`* number we are looking for is higher than anything currently assigned`
`679`	`679`	`* and there is already an assert above for this case.`
`680`	`680`	`*/`
`681`		`- LOG(EXTENSION_LOG_WARNING, "Cursor not registered into vb %d "`
`682`		`- " for stream '%s' because seqno %" PRIu64 " is too high",`
`683`		`- vbucketId, name.c_str(), startBySeqno);`
	`681`	`+ throw std::logic_error(`
	`682`	`+ "CheckpointManager::registerCursorBySeqno the sequences number "`
	`683`	`+ "is higher than anything currently assigned");`
`684`	`684`	`}`
`685`	`685`	`return result;`
`686`	`686`	`}`
Original file line number	Diff line number	Diff line change
`@@ -86,4 +86,8 @@ class MockActiveStream : public ActiveStream {`
`86`	`86`	`bool public_getPendingBackfill() const {`
`87`	`87`	`return pendingBackfill;`
`88`	`88`	`}`
	`89`	`+`
	`90`	`+ void transitionStateToBackfilling() {`
	`91`	`+ transitionState(STREAM_BACKFILLING);`
	`92`	`+ }`
`89`	`93`	`};`