MB-29369: Obtain the streamMutex earlier in the snapshot task

Obtain the streamMutex and also check the stream is in-memory /
takeover-send before the task increments the cursor.

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

Author: jimwwalker

engines/ep/src/dcp/stream.cc

@@ -1075,10 +1075,20 @@ void ActiveStreamCheckpointProcessorTask::addStats(const std::string& name,
 }
 
 void ActiveStream::nextCheckpointItemTask() {
+    // MB-29369: Obtain stream mutex here
+    LockHolder lh(streamMutex);
+    nextCheckpointItemTask(lh);
+}
+
+void ActiveStream::nextCheckpointItemTask(const LockHolder& streamMutex) {
     VBucketPtr vbucket = engine->getVBucket(vb_);
     if (vbucket) {
-        auto items = getOutstandingItems(*vbucket);
-        processItems(items);
+        // MB-29369: only run the task's work if the stream is in an in-memory
+        // phase (of which takeover is a variant).
+        if (isInMemory() || isTakeoverSend()) {
+            auto items = getOutstandingItems(*vbucket);
+            processItems(items, streamMutex);
+        }
     } else {
         /* The entity deleting the vbucket must set stream to dead,
            calling setDead(END_STREAM_STATE) will cause deadlock because
@@ -1207,7 +1217,8 @@ std::unique_ptr<DcpResponse> ActiveStream::makeResponseFromItem(
     }
 }
 
-void ActiveStream::processItems(std::vector<queued_item>& items) {
+void ActiveStream::processItems(std::vector<queued_item>& items,
+                                const LockHolder& streamMutex) {
     if (!items.empty()) {
         bool mark = false;
         if (items.front()->getOperation() == queue_op::checkpoint_start) {
@@ -1243,7 +1254,7 @@ void ActiveStream::processItems(std::vector<queued_item>& items) {
         if (mutations.empty()) {
             // If we only got checkpoint start or ends check to see if there are
             // any more snapshots before pausing the stream.
-            nextCheckpointItemTask();
+            nextCheckpointItemTask(streamMutex);
         } else {
             snapshot(mutations, mark);
         }
@@ -1267,19 +1278,6 @@ void ActiveStream::snapshot(std::deque<std::unique_ptr<DcpResponse>>& items,
         return;
     }
 
-    LockHolder lh(streamMutex);
-
-    if (!isActive() || isBackfilling()) {
-        // If stream was closed forcefully by the time the checkpoint items
-        // retriever task completed, or if we decided to switch the stream to
-        // backfill state from in-memory state, none of the acquired mutations
-        // should be added on the stream's readyQ. We must drop items in case
-        // we switch state from in-memory to backfill because we schedule
-        // backfill from lastReadSeqno + 1
-        items.clear();
-        return;
-    }
-
     /* This assumes that all items in the "items deque" is put onto readyQ */
     lastReadSeqno.store(lastReadSeqnoUnSnapshotted);
 

engines/ep/src/dcp/stream.h

@@ -322,7 +322,8 @@ class ActiveStream : public Stream,
 
     // Given a set of queued items, create mutation responses for each item,
     // and pass onto the producer associated with this stream.
-    void processItems(std::vector<queued_item>& items);
+    void processItems(std::vector<queued_item>& items,
+                      const LockHolder& streamMutex);
 
     bool nextCheckpointItem();
 
@@ -359,6 +360,12 @@ class ActiveStream : public Stream,
     virtual void registerCursor(CheckpointManager& chkptmgr,
                                 uint64_t lastProcessedSeqno);
 
+    /**
+     * Unlocked variant of nextCheckpointItemTask caller must obtain
+     * streamMutex and pass a reference to it
+     * @param streamMutex reference to lockholder
+     */
+    void nextCheckpointItemTask(const LockHolder& streamMutex);
 
     /* Indicates that a backfill has been scheduled and has not yet completed.
      * Is protected (as opposed to private) for testing purposes.

engines/ep/tests/mock/mock_dcp_producer.cc

@@ -19,14 +19,15 @@
 
 #include "mock_stream.h"
 
-void MockDcpProducer::mockActiveStreamRequest(uint32_t flags,
-                                              uint32_t opaque,
-                                              VBucket& vb,
-                                              uint64_t start_seqno,
-                                              uint64_t end_seqno,
-                                              uint64_t vbucket_uuid,
-                                              uint64_t snap_start_seqno,
-                                              uint64_t snap_end_seqno) {
+std::shared_ptr<MockActiveStream> MockDcpProducer::mockActiveStreamRequest(
+        uint32_t flags,
+        uint32_t opaque,
+        VBucket& vb,
+        uint64_t start_seqno,
+        uint64_t end_seqno,
+        uint64_t vbucket_uuid,
+        uint64_t snap_start_seqno,
+        uint64_t snap_end_seqno) {
     auto stream = std::make_shared<MockActiveStream>(
             static_cast<EventuallyPersistentEngine*>(&engine_),
             std::static_pointer_cast<MockDcpProducer>(shared_from_this()),
@@ -45,4 +46,5 @@ void MockDcpProducer::mockActiveStreamRequest(uint32_t flags,
                 "failed to insert requested stream");
     }
     notifyStreamReady(vb.getId());
+    return stream;
 }

engines/ep/tests/mock/mock_dcp_producer.h

@@ -23,6 +23,8 @@
 #include "kv_bucket.h"
 #include "mock_dcp_backfill_mgr.h"
 
+class MockActiveStream;
+
 /*
  * Mock of the DcpProducer class.  Wraps the real DcpProducer, but exposes
  * normally protected methods publically for test purposes.
@@ -147,12 +149,13 @@ class MockDcpProducer : public DcpProducer {
     /**
      * Place a mock active stream into the producer
      */
-    void mockActiveStreamRequest(uint32_t flags,
-                                 uint32_t opaque,
-                                 VBucket& vb,
-                                 uint64_t start_seqno,
-                                 uint64_t end_seqno,
-                                 uint64_t vbucket_uuid,
-                                 uint64_t snap_start_seqno,
-                                 uint64_t snap_end_seqno);
+    std::shared_ptr<MockActiveStream> mockActiveStreamRequest(
+            uint32_t flags,
+            uint32_t opaque,
+            VBucket& vb,
+            uint64_t start_seqno,
+            uint64_t end_seqno,
+            uint64_t vbucket_uuid,
+            uint64_t snap_start_seqno,
+            uint64_t snap_end_seqno);
 };

engines/ep/tests/mock/mock_stream.h

@@ -65,7 +65,8 @@ class MockActiveStream : public ActiveStream {
     }
 
     void public_processItems(std::vector<queued_item>& items) {
-        processItems(items);
+        LockHolder lh(streamMutex);
+        processItems(items, lh);
     }
 
     bool public_nextCheckpointItem() {

engines/ep/tests/module_tests/dcp_test.cc

@@ -135,6 +135,7 @@ class DCPTest : public EventuallyPersistentEngineTest {
                 stream->getCursorName(), 1, false, MustSendCheckpointEnd::NO))
                 << "Found an existing TAP cursor when attempting to register "
                    "ours";
+        stream->setActive();
     }
 
     void destroy_dcp_stream() {
@@ -1104,36 +1105,6 @@ TEST_P(StreamTest, MB17653_ItemsRemaining) {
     destroy_dcp_stream();
 }
 
-TEST_P(StreamTest, test_mb18625) {
-    // Add an item.
-    store_item(vbid, "key", "value");
-
-    setup_dcp_stream();
-
-    // Should start with nextCheckpointItem() returning true.
-    EXPECT_TRUE(stream->public_nextCheckpointItem())
-            << "nextCheckpointItem() should initially be true.";
-
-    // Get the set of outstanding items
-    auto items = stream->public_getOutstandingItems(*vb0);
-
-    // Set stream to DEAD to simulate a close stream request
-    stream->setDead(END_STREAM_CLOSED);
-
-    // Process the set of items retrieved from checkpoint queues previously
-    stream->public_processItems(items);
-
-    // Retrieve the next message in the stream's readyQ
-    auto op = stream->public_nextQueuedItem();
-    EXPECT_EQ(DcpResponse::Event::StreamEnd, op->getEvent())
-        << "Expected the STREAM_END message";
-
-    // Expect no other message to be queued after stream end message
-    EXPECT_EQ(0, (stream->public_readyQ()).size())
-            << "Expected no more messages in the readyQ";
-    destroy_dcp_stream();
-}
-
 /* Stream items from a DCP backfill */
 TEST_P(StreamTest, BackfillOnly) {
     /* Add 3 items */
@@ -1372,7 +1343,6 @@ TEST_P(StreamTest, CursorDroppingBasicNotAllowedStates) {
 
     /* Transition stream to takeoverSend state and expect cursor dropping call
        to fail */
-    stream->transitionStateToBackfilling();
     stream->transitionStateToTakeoverSend();
     EXPECT_FALSE(stream->public_handleSlowStream());