Merge branch 'couchbase/spock' into 'couchbase/vulcan'

* 'couchbase/spock':
  MB-29381: Allow DCP rollback on vbuckets in pending state
  MB-29287: Test that takeover finds all items
  MB-29287: Give each ActiveStream a unique cursor name

Change-Id: I98a2fdde67c2c003504da8219112434e1f2f1b0a

Author: jimwwalker

engines/ep/CMakeLists.txt

@@ -268,6 +268,7 @@ if (COUCHBASE_KV_BUILD_UNIT_TESTS)
     # so simpler / quicker just to link them into a single executable).
     ADD_EXECUTABLE(ep-engine_ep_unit_tests
                    tests/mock/mock_dcp.cc
+                   tests/mock/mock_dcp_producer.cc
                    tests/mock/mock_ephemeral_vb.cc
                    tests/mock/mock_synchronous_ep_engine.cc
                    tests/module_tests/atomic_unordered_map_test.cc

engines/ep/src/dcp/producer.cc

@@ -1047,12 +1047,12 @@ void DcpProducer::closeStreamDueToRollback(uint16_t vbucket) {
 }
 
 bool DcpProducer::handleSlowStream(uint16_t vbid,
-                                   const std::string &name) {
+                                   const std::string& cursorName) {
     if (supportsCursorDropping) {
         auto stream = findStream(vbid);
         if (stream) {
-            if (stream->getName().compare(name) == 0) {
-                ActiveStream* as = static_cast<ActiveStream*>(stream.get());
+            ActiveStream* as = static_cast<ActiveStream*>(stream.get());
+            if (as->getCursorName() == cursorName) {
                 return as->handleSlowStream();
             }
         }

engines/ep/src/dcp/producer.h

@@ -101,10 +101,17 @@ class DcpProducer : public ConnHandler,
 
     void closeStreamDueToRollback(uint16_t vbucket);
 
-    /* This function handles a stream that is detected as slow by the checkpoint
-       remover. Currently we handle the slow stream by switching from in-memory
-       to backfilling */
-    bool handleSlowStream(uint16_t vbid, const std::string &name);
+    /**
+     * This function handles a stream that is detected as slow by the checkpoint
+     * remover. Currently we handle the slow stream by switching from in-memory
+     * to backfilling.
+     *
+     * @param vbid vbucket the checkpoint-remover is processing
+     * @param cursorName the cursor name registered in the checkpoint manager
+     *        which is slow.
+     * @return true if the cursor was removed from the checkpoint manager
+     */
+    bool handleSlowStream(uint16_t vbid, const std::string& cursorName);
 
     void closeAllStreams();
 

engines/ep/src/dcp/stream.cc

@@ -271,7 +271,8 @@ ActiveStream::ActiveStream(EventuallyPersistentEngine* e,
       includeValue(includeVal),
       includeXattributes(includeXattrs),
       includeDeleteTime(includeDeleteTime),
-      filter(filter, manifest) {
+      filter(filter, manifest),
+      cursorName(n + std::to_string(cursorUID.fetch_add(1))) {
     const char* type = "";
     if (flags_ & DCP_ADD_STREAM_FLAG_TAKEOVER) {
         type = "takeover ";
@@ -376,7 +377,7 @@ void ActiveStream::registerCursor(CheckpointManager& chkptmgr,
                                   uint64_t lastProcessedSeqno) {
     try {
         CursorRegResult result = chkptmgr.registerCursorBySeqno(
-                name_, lastProcessedSeqno, MustSendCheckpointEnd::NO);
+                cursorName, lastProcessedSeqno, MustSendCheckpointEnd::NO);
         /*
          * MB-22960:  Due to cursor dropping we re-register the replication
          * cursor only during backfill when we mark the disk snapshot.  However
@@ -885,8 +886,9 @@ void ActiveStream::addTakeoverStats(ADD_STAT add_stat, const void *cookie,
 
     size_t vb_items = vb.getNumItems();
     size_t chk_items =
-            vb_items > 0 ? vb.checkpointManager->getNumItemsForCursor(name_)
-                         : 0;
+            vb_items > 0
+                    ? vb.checkpointManager->getNumItemsForCursor(cursorName)
+                    : 0;
 
     size_t del_items = 0;
     try {
@@ -945,7 +947,7 @@ std::unique_ptr<DcpResponse> ActiveStream::nextQueuedItem() {
 bool ActiveStream::nextCheckpointItem() {
     VBucketPtr vbucket = engine->getVBucket(vb_);
     if (vbucket &&
-        vbucket->checkpointManager->getNumItemsForCursor(name_) > 0) {
+        vbucket->checkpointManager->getNumItemsForCursor(cursorName) > 0) {
         // schedule this stream to build the next checkpoint
         auto producer = producerPtr.lock();
         if (!producer) {
@@ -1092,7 +1094,7 @@ std::vector<queued_item> ActiveStream::getOutstandingItems(VBucket& vb) {
     chkptItemsExtractionInProgress.store(true);
 
     auto _begin_ = ProcessClock::now();
-    vb.checkpointManager->getAllItemsForCursor(name_, items);
+    vb.checkpointManager->getAllItemsForCursor(cursorName, items);
     engine->getEpStats().dcpCursorsGetItemsHisto.add(
             std::chrono::duration_cast<std::chrono::microseconds>(
                     ProcessClock::now() - _begin_));
@@ -1451,7 +1453,7 @@ void ActiveStream::scheduleBackfill_UNLOCKED(bool reschedule) {
         try {
             std::tie(curChkSeqno, tryBackfill) =
                     vbucket->checkpointManager->registerCursorBySeqno(
-                            name_,
+                            cursorName,
                             lastReadSeqno.load(),
                             MustSendCheckpointEnd::NO);
         } catch(std::exception& error) {
@@ -1553,7 +1555,7 @@ void ActiveStream::scheduleBackfill_UNLOCKED(bool reschedule) {
             try {
                 CursorRegResult result =
                         vbucket->checkpointManager->registerCursorBySeqno(
-                                name_,
+                                cursorName,
                                 lastReadSeqno.load(),
                                 MustSendCheckpointEnd::NO);
 
@@ -1777,7 +1779,7 @@ size_t ActiveStream::getItemsRemaining() {
     // Items remaining is the sum of:
     // (a) Items outstanding in checkpoints
     // (b) Items pending in our readyQ, excluding any meta items.
-    return vbucket->checkpointManager->getNumItemsForCursor(name_) +
+    return vbucket->checkpointManager->getNumItemsForCursor(cursorName) +
            readyQ_non_meta_items;
 }
 
@@ -1826,7 +1828,7 @@ bool ActiveStream::dropCheckpointCursor_UNLOCKED() {
         notifyStreamReady();
     }
     /* Drop the existing cursor */
-    return vbucket->checkpointManager->removeCursor(name_);
+    return vbucket->checkpointManager->removeCursor(cursorName);
 }
 
 EXTENSION_LOG_LEVEL ActiveStream::getTransitionStateLogLevel(
@@ -1864,10 +1866,12 @@ void ActiveStream::notifyStreamReady(bool force) {
 void ActiveStream::removeCheckpointCursor() {
     VBucketPtr vb = engine->getVBucket(vb_);
     if (vb) {
-        vb->checkpointManager->removeCursor(name_);
+        vb->checkpointManager->removeCursor(cursorName);
     }
 }
 
+std::atomic<uint64_t> ActiveStream::cursorUID;
+
 NotifierStream::NotifierStream(EventuallyPersistentEngine* e,
                                std::shared_ptr<DcpProducer> p,
                                const std::string& name,

engines/ep/src/dcp/stream.h

@@ -307,13 +307,18 @@ class ActiveStream : public Stream,
         return currentSeparator;
     }
 
+    /// @return a const reference to the streams cursor name
+    const std::string& getCursorName() const {
+        return cursorName;
+    }
+
 protected:
     /**
      * @param vb reference to the associated vbucket
      *
      * @return the outstanding items for the stream's checkpoint cursor.
      */
-    std::vector<queued_item> getOutstandingItems(VBucket& vb);
+    virtual std::vector<queued_item> getOutstandingItems(VBucket& vb);
 
     // Given a set of queued items, create mutation responses for each item,
     // and pass onto the producer associated with this stream.
@@ -516,6 +521,17 @@ class ActiveStream : public Stream,
      * The filter the stream will use to decide which keys should be transmitted
      */
     Collections::VB::Filter filter;
+
+    /**
+     * The name which uniquely identifies this stream's checkpoint cursor
+     */
+    std::string cursorName;
+
+    /**
+     * To ensure each stream gets a unique cursorName, we maintain a 'uid'
+     * which is really just an incrementing uint64
+     */
+    static std::atomic<uint64_t> cursorUID;
 };
 
 

engines/ep/tests/mock/mock_dcp_producer.cc

@@ -0,0 +1,48 @@
+/* -*- 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.
+ */
+
+#include "mock_dcp_producer.h"
+
+#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) {
+    auto stream = std::make_shared<MockActiveStream>(
+            static_cast<EventuallyPersistentEngine*>(&engine_),
+            std::static_pointer_cast<MockDcpProducer>(shared_from_this()),
+            flags,
+            opaque,
+            vb,
+            start_seqno,
+            end_seqno,
+            vbucket_uuid,
+            snap_start_seqno,
+            snap_end_seqno);
+    stream->setActive();
+    if (!streams.insert(std::make_pair(vb.getId(), stream))) {
+        throw std::logic_error(
+                "MockDcpProducer::mockActiveStreamRequest "
+                "failed to insert requested stream");
+    }
+    notifyStreamReady(vb.getId());
+}

engines/ep/tests/mock/mock_dcp_producer.h

@@ -143,4 +143,16 @@ class MockDcpProducer : public DcpProducer {
     size_t getBytesOutstanding() const {
         return log.getBytesOutstanding();
     }
+
+    /**
+     * 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);
 };

engines/ep/tests/mock/mock_stream.h

@@ -18,6 +18,8 @@
 #pragma once
 
 #include "collections/vbucket_filter.h"
+#include "dcp/consumer.h"
+#include "dcp/producer.h"
 #include "dcp/stream.h"
 #include "tests/mock/mock_dcp_producer.h"
 
@@ -147,6 +149,14 @@ class MockActiveStream : public ActiveStream {
     void setState(StreamState state) {
         state_ = state;
     }
+
+    virtual std::vector<queued_item> getOutstandingItems(VBucket& vb) override {
+        preGetOutstandingItemsCallback();
+        return ActiveStream::getOutstandingItems(vb);
+    }
+
+    /// A callback to allow tests to inject code before we access the checkpoint
+    std::function<void()> preGetOutstandingItemsCallback = [] { return; };
 };
 
 /**

engines/ep/tests/module_tests/dcp_test.cc

@@ -132,7 +132,7 @@ class DCPTest : public EventuallyPersistentEngineTest {
                                                     includeXattrs);
 
         EXPECT_FALSE(vb0->checkpointManager->registerCursor(
-                producer->getName(), 1, false, MustSendCheckpointEnd::NO))
+                stream->getCursorName(), 1, false, MustSendCheckpointEnd::NO))
                 << "Found an existing TAP cursor when attempting to register "
                    "ours";
     }
@@ -3225,21 +3225,8 @@ class ActiveStreamChkptProcessorTaskTest : public SingleThreadedKVBucketTest {
     }
 
     void notifyAndStepToCheckpoint() {
-        auto vb = store->getVBucket(vbid);
-        ASSERT_NE(nullptr, vb.get());
-
-        producer->notifySeqnoAvailable(vbid, vb->getHighSeqno());
-
-        /* Step which will notify the checkpoint processor task */
-        EXPECT_EQ(ENGINE_SUCCESS, producer->step(producers.get()));
-        EXPECT_EQ(1, producer->getCheckpointSnapshotTask().queueSize());
-
-        /* Run the task */
-        producer->getCheckpointSnapshotTask().run();
-
-        /* This time the step should return something that is read from the
-           checkpoint processor */
-        EXPECT_EQ(ENGINE_WANT_MORE, producer->step(producers.get()));
+        SingleThreadedKVBucketTest::notifyAndStepToCheckpoint(*producer,
+                                                              *producers);
     }
 
     const void* cookie;