[BP] MB-25798: Backfill task leave stream state unchanged

The backfill task must not change the state machine of the active
stream that it is associated with.  In particular:
1) In ActiveStream::markDiskSnapshot if a vbucket is not found then
   it should just return, leaving the entity deleting the vbucket to
   set the stream to dead.
2) ActiveStream::completeBackfill should not call
   scheduleBackfill_UNLOCKED as this can move the stream into the
   STREAM_IN_MEMORY state.

State machine changes should only be driven by ActiveStream::next,
which is invoked by DcpProducer::getNextItem.

The call to scheduleBackfill_UNLOCKED that was invoked in
ActiveStream::completeBackfill when the pendingBackfill flag was true
has been moved to the ActiveStream::backfillPhase function and is
invoked once the current backfill has completed and the pendingBackfill
flag is set to true.

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

Author: owend74

src/dcp/stream.cc

@@ -238,7 +238,7 @@ DcpResponse* ActiveStream::next() {
             break;
         case STREAM_BACKFILLING:
             validTransition = true;
-            response = backfillPhase();
+            response = backfillPhase(lh);
             break;
         case STREAM_IN_MEMORY:
             validTransition = true;
@@ -283,16 +283,26 @@ void ActiveStream::markDiskSnapshot(uint64_t startSeqno, uint64_t endSeqno) {
     uint64_t chkCursorSeqno = endSeqno;
 
     if (state_ != STREAM_BACKFILLING) {
+        producer->getLogger().log(EXTENSION_LOG_WARNING,
+                                  "(vb %" PRIu16 ") ActiveStream::"
+                                  "markDiskSnapshot: Unexpected state_:%s",
+                                  vb_, stateName(state_));
         return;
     }
 
     startSeqno = std::min(snap_start_seqno_, startSeqno);
     firstMarkerSent = true;
 
     RCPtr<VBucket> vb = engine->getVBucket(vb_);
+    if (!vb) {
+        producer->getLogger().log(EXTENSION_LOG_WARNING,"(vb %" PRIu16 ") "
+                                  "ActiveStream::markDiskSnapshot, vbucket "
+                                  "does not exist", vb_);
+        return;
+    }
     // An atomic read of vbucket state without acquiring the
     // reader lock for state should suffice here.
-    if (vb && vb->getState() == vbucket_state_replica) {
+    if (vb->getState() == vbucket_state_replica) {
         if (end_seqno_ > endSeqno) {
             /* We possibly have items in the open checkpoint
                (incomplete snapshot) */
@@ -314,9 +324,7 @@ void ActiveStream::markDiskSnapshot(uint64_t startSeqno, uint64_t endSeqno) {
                                     MARKER_FLAG_DISK));
     lastSentSnapEndSeqno.store(endSeqno, std::memory_order_relaxed);
 
-    if (!vb) {
-        endStream(END_STREAM_STATE);
-    } else if (!(flags_ & DCP_ADD_STREAM_FLAG_DISKONLY)) {
+    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(
@@ -384,26 +392,10 @@ void ActiveStream::completeBackfill() {
                     uint64_t(backfillItems.memory.load()),
                     lastReadSeqno.load(),
                     pendingBackfill ? "True" : "False");
-
-            isBackfillTaskRunning = false;
-            if (pendingBackfill) {
-                scheduleBackfill_UNLOCKED(true);
-                pendingBackfill = false;
-            }
-
-            bool expected = false;
-            if (itemsReady.compare_exchange_strong(expected, true)) {
-                producer->notifyStreamReady(vb_);
-            }
-
-            /**
-              * MB-22451: It is important that we return here because
-              * scheduleBackfill_UNLOCKED(true) can set
-              * isBackfillTaskRunning to true.  Therefore if we don't return we
-              * will set isBackfillTaskRunning prematurely back to false, (see
-              * below).
-              */
-            return;
+        } else {
+            producer->getLogger().log(EXTENSION_LOG_WARNING,
+                    "(vb %" PRIu16 ") ActiveStream::completeBackfill: "
+                    "Unexpected state_:%s", vb_, stateName(state_));
         }
     }
 
@@ -488,7 +480,7 @@ void ActiveStream::setVBucketStateAckRecieved() {
     }
 }
 
-DcpResponse* ActiveStream::backfillPhase() {
+DcpResponse* ActiveStream::backfillPhase(LockHolder& lh) {
     DcpResponse* resp = nextQueuedItem();
 
     if (resp && (resp->getEvent() == DCP_MUTATION ||
@@ -504,19 +496,27 @@ DcpResponse* ActiveStream::backfillPhase() {
     }
 
     if (!isBackfillTaskRunning && readyQ.empty()) {
+        // Given readyQ.empty() is True resp will be NULL
         backfillRemaining.store(0, std::memory_order_relaxed);
-        if (lastReadSeqno.load() >= end_seqno_) {
-            endStream(END_STREAM_OK);
-        } else if (flags_ & DCP_ADD_STREAM_FLAG_TAKEOVER) {
-            transitionState(STREAM_TAKEOVER_SEND);
-        } else if (flags_ & DCP_ADD_STREAM_FLAG_DISKONLY) {
-            endStream(END_STREAM_OK);
+        // The previous backfill has completed.  Check to see if another
+        // backfill needs to be scheduled.
+        if (pendingBackfill) {
+            scheduleBackfill_UNLOCKED(true);
+            pendingBackfill = false;
         } else {
-            transitionState(STREAM_IN_MEMORY);
-        }
+            if (lastReadSeqno.load() >= end_seqno_) {
+                endStream(END_STREAM_OK);
+            } else if (flags_ & DCP_ADD_STREAM_FLAG_TAKEOVER) {
+                transitionState(STREAM_TAKEOVER_SEND);
+            } else if (flags_ & DCP_ADD_STREAM_FLAG_DISKONLY) {
+                endStream(END_STREAM_OK);
+            } else {
+                transitionState(STREAM_IN_MEMORY);
+            }
 
-        if (!resp) {
-            resp = nextQueuedItem();
+            if (!resp) {
+                resp = nextQueuedItem();
+            }
         }
     }
 
@@ -528,6 +528,8 @@ DcpResponse* ActiveStream::inMemoryPhase() {
         endStream(END_STREAM_OK);
     } else if (readyQ.empty()) {
         if (pendingBackfill) {
+            // Moving the state from STREAM_IN_MEMORY to
+            // STREAM_BACKFILLING will result in a backfill being scheduled
             transitionState(STREAM_BACKFILLING);
             pendingBackfill = false;
             return NULL;

src/dcp/stream.h

@@ -281,7 +281,7 @@ class ActiveStream : public Stream {
 
 private:
 
-    DcpResponse* backfillPhase();
+    DcpResponse* backfillPhase(LockHolder& lh);
 
     DcpResponse* inMemoryPhase();
 

tests/module_tests/evp_store_single_threaded_test.cc

@@ -123,8 +123,9 @@ class SingleThreadedEPStoreTest : public EventuallyPersistentStoreTest {
 /**
  * Regression test for MB-22451: When handleSlowStream is called and in
  * STREAM_BACKFILLING state and currently have a backfill scheduled (or running)
- * ensure that when the backfill completes the new backfill is scheduled and
- * the backfilling flag remains true.
+ * ensure that when the backfill completes pendingBackfill remains true,
+ * isBackfillTaskRunning is false and, the stream state remains set to
+ * STREAM_BACKFILLING.
  */
 TEST_F(SingleThreadedEPStoreTest, test_mb22451) {
     // Make vbucket active.
@@ -168,11 +169,12 @@ TEST_F(SingleThreadedEPStoreTest, test_mb22451) {
     // The call to handleSlowStream should result in setting pendingBackfill
     // flag to true
     EXPECT_TRUE(mock_stream->public_getPendingBackfill())
-        << "pendingBackfill is not true";
+        << "handleSlowStream should set pendingBackfill to True";
     mock_stream->completeBackfill();
-    EXPECT_TRUE(mock_stream->public_isBackfillTaskRunning())
-        << "isBackfillRunning is not true";
-
+    EXPECT_FALSE(mock_stream->public_isBackfillTaskRunning())
+        << "completeBackfill should set isBackfillTaskRunning to False";
+    EXPECT_EQ(STREAM_BACKFILLING, mock_stream->getState())
+            << "stream state should not have changed";
     // Required to ensure that the backfillMgr is deleted
     producer->closeAllStreams();
 }