Merge 'ep-engine/watson' into 'kv-engine/watson_ep'

* ep-engine/watson:
  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: I8753ecdaa8ebd5f52f3c319542da346e3eaeb45c

Author: jimwwalker

CMakeLists.txt

@@ -185,6 +185,7 @@ TARGET_LINK_LIBRARIES(ep cJSON JSON_checker couchstore forestdb
 # 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/module_tests/checkpoint_test.cc
   tests/module_tests/defragmenter_test.cc
   tests/module_tests/ep_unit_tests_main.cc

src/dcp/producer.cc

@@ -799,12 +799,12 @@ void DcpProducer::vbucketStateChanged(uint16_t vbucket, vbucket_state_t state) {
 }
 
 bool DcpProducer::handleSlowStream(uint16_t vbid,
-                                   const std::string &name) {
+                                   const std::string& cursorName) {
     if (supportsCursorDropping) {
         stream_t stream = findStreamByVbid(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) {
                 as->handleSlowStream();
                 return true;
             }

src/dcp/producer.h

@@ -72,10 +72,17 @@ class DcpProducer : public Producer {
 
     void vbucketStateChanged(uint16_t vbucket, vbucket_state_t state);
 
-    /* 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();
 
@@ -230,7 +237,7 @@ class DcpProducer : public Producer {
 
     ExTask checkpointCreatorTask;
 
-private:
+protected:
 
 
     DcpResponse* getNextItem();

src/dcp/stream.cc

@@ -182,8 +182,8 @@ ActiveStream::ActiveStream(EventuallyPersistentEngine* e, dcp_producer_t p,
        engine(e), producer(p),
        payloadType((flags & DCP_ADD_STREAM_FLAG_NO_VALUE) ? KEY_ONLY :
                                                             KEY_VALUE),
-       lastSentSnapEndSeqno(0), chkptItemsExtractionInProgress(false) {
-
+       lastSentSnapEndSeqno(0), chkptItemsExtractionInProgress(false),
+       cursorName(n + std::to_string(cursorUID.fetch_add(1))) {
     const char* type = "";
     if (flags_ & DCP_ADD_STREAM_FLAG_TAKEOVER) {
         type = "takeover ";
@@ -338,7 +338,7 @@ void ActiveStream::markDiskSnapshot(uint64_t startSeqno, uint64_t endSeqno) {
         try {
             CursorRegResult result =
                     vb->checkpointManager.registerCursorBySeqno(
-                            name_, chkCursorSeqno,
+                            cursorName, chkCursorSeqno,
                             MustSendCheckpointEnd::NO);
 
             curChkSeqno = result.first;
@@ -696,7 +696,7 @@ void ActiveStream::addTakeoverStats(ADD_STAT add_stat, const void *cookie) {
     item_eviction_policy_t iep = engine->getEpStore()->getItemEvictionPolicy();
     size_t vb_items = vb->getNumItems(iep);
     size_t chk_items = vb_items > 0 ?
-                vb->checkpointManager.getNumItemsForCursor(name_) : 0;
+                vb->checkpointManager.getNumItemsForCursor(cursorName) : 0;
 
     size_t del_items = 0;
     try {
@@ -747,7 +747,7 @@ DcpResponse* ActiveStream::nextQueuedItem() {
 
 bool ActiveStream::nextCheckpointItem() {
     RCPtr<VBucket> vbucket = engine->getVBucket(vb_);
-    if (vbucket && vbucket->checkpointManager.getNumItemsForCursor(name_) > 0) {
+    if (vbucket && vbucket->checkpointManager.getNumItemsForCursor(cursorName) > 0) {
         // schedule this stream to build the next checkpoint
         producer->scheduleCheckpointProcessorTask(this);
         return true;
@@ -837,7 +837,7 @@ void ActiveStream::getOutstandingItems(RCPtr<VBucket> &vb,
     // Commencing item processing - set guard flag.
     chkptItemsExtractionInProgress.store(true);
 
-    vb->checkpointManager.getAllItemsForCursor(name_, items);
+    vb->checkpointManager.getAllItemsForCursor(cursorName, items);
     if (vb->checkpointManager.getNumCheckpoints() > 1) {
         engine->getEpStore()->wakeUpCheckpointRemover();
     }
@@ -1047,7 +1047,8 @@ void ActiveStream::scheduleBackfill_UNLOCKED(bool reschedule) {
         try {
             std::tie(curChkSeqno, tryBackfill) =
                     vbucket->checkpointManager.registerCursorBySeqno(
-                            name_, lastReadSeqno.load(),
+                            cursorName,
+                            lastReadSeqno.load(),
                             MustSendCheckpointEnd::NO);
         } catch(std::exception& error) {
             producer->getLogger().log(EXTENSION_LOG_WARNING,
@@ -1125,11 +1126,12 @@ void ActiveStream::scheduleBackfill_UNLOCKED(bool reschedule) {
              */
             try {
                 CursorRegResult result =
-                            vbucket->checkpointManager.registerCursorBySeqno(
-                            name_, lastReadSeqno.load(),
-                            MustSendCheckpointEnd::NO);
+                        vbucket->checkpointManager.registerCursorBySeqno(
+                                cursorName,
+                                lastReadSeqno.load(),
+                                MustSendCheckpointEnd::NO);
 
-                    curChkSeqno = result.first;
+                curChkSeqno = result.first;
             } catch (std::exception& error) {
                 producer->getLogger().log(EXTENSION_LOG_WARNING,
                                           "(vb %" PRIu16 ") Failed to register "
@@ -1307,7 +1309,7 @@ void ActiveStream::transitionState(stream_state_t newState) {
             {
                 RCPtr<VBucket> vb = engine->getVBucket(vb_);
                 if (vb) {
-                    vb->checkpointManager.removeCursor(name_);
+                    vb->checkpointManager.removeCursor(cursorName);
                 }
                 break;
             }
@@ -1330,8 +1332,8 @@ 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_) +
-            readyQ_non_meta_items;
+    return vbucket->checkpointManager.getNumItemsForCursor(cursorName) +
+           readyQ_non_meta_items;
 }
 
 uint64_t ActiveStream::getLastReadSeqno() const {
@@ -1377,9 +1379,11 @@ void ActiveStream::dropCheckpointCursor_UNLOCKED()
         }
     }
     /* Drop the existing cursor */
-    vbucket->checkpointManager.removeCursor(name_);
+    vbucket->checkpointManager.removeCursor(cursorName);
 }
 
+std::atomic<uint64_t> ActiveStream::cursorUID;
+
 NotifierStream::NotifierStream(EventuallyPersistentEngine* e, dcp_producer_t p,
                                const std::string &name, uint32_t flags,
                                uint32_t opaque, uint16_t vb, uint64_t st_seqno,

src/dcp/stream.h

@@ -257,9 +257,14 @@ class ActiveStream : public Stream {
        in-memory to backfilling */
     void handleSlowStream();
 
+    /// @return a const reference to the streams cursor name
+    const std::string& getCursorName() const {
+        return cursorName;
+    }
+
 protected:
     // Returns the outstanding items for the stream's checkpoint cursor.
-    void getOutstandingItems(RCPtr<VBucket> &vb, std::vector<queued_item> &items);
+    virtual void getOutstandingItems(RCPtr<VBucket> &vb, std::vector<queued_item> &items);
 
     // Given a set of queued items, create mutation responses for each item,
     // and pass onto the producer associated with this stream.
@@ -381,6 +386,17 @@ class ActiveStream : public Stream {
        items are added to the readyQ */
     AtomicValue<bool> chkptItemsExtractionInProgress;
 
+    /**
+     * 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;
+
 };
 
 

src/ep.cc

@@ -4008,7 +4008,8 @@ TaskStatus EventuallyPersistentStore::rollback(uint16_t vbid,
     }
 
     ReaderLockHolder rlh(vb->getStateLock());
-    if (vb->getState() == vbucket_state_replica) {
+    if ((vb->getState() == vbucket_state_replica) ||
+        (vb->getState() == vbucket_state_pending)) {
         uint64_t prevHighSeqno =
                 static_cast<uint64_t>(vb->checkpointManager.getHighSeqno());
         if (rollbackSeqno != 0) {

tests/mock/mock_dcp_producer.cc

@@ -0,0 +1,49 @@
+/* -*- 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,
+                                              uint16_t vbucket,
+                                              uint64_t start_seqno,
+                                              uint64_t end_seqno,
+                                              uint64_t vbucket_uuid,
+                                              uint64_t snap_start_seqno,
+                                              uint64_t snap_end_seqno) {
+    stream_t stream = new MockActiveStream(
+            static_cast<EventuallyPersistentEngine*>(&engine_),
+            this,
+            getName(),
+            flags,
+            opaque,
+            vbucket,
+            start_seqno,
+            end_seqno,
+            vbucket_uuid,
+            snap_start_seqno,
+            snap_end_seqno);
+    stream->setActive();
+    auto insertResult = streams.insert(std::make_pair(vbucket, stream));
+    if (!insertResult.second) {
+        throw std::logic_error("MockDcpProducer::mockActiveStreamRequest "
+                               "failed to insert requested stream");
+    }
+    notifyStreamReady(vbucket);
+}

tests/mock/mock_dcp_producer.h

@@ -60,6 +60,30 @@ class MockDcpProducer: public DcpProducer {
         return *static_cast<ActiveStreamCheckpointProcessorTask*>(
                         checkpointCreatorTask.get());
     }
+
+    /**
+     * Finds the stream for a given vbucket
+     */
+    SingleThreadedRCPtr<Stream> findStream(uint16_t vbid) {
+        auto it = streams.find(vbid);
+        if (it != streams.end()) {
+            return it->second;
+        } else {
+            return SingleThreadedRCPtr<Stream>();
+        }
+    }
+
+    /**
+     * Place a mock active stream into the producer
+     */
+    void mockActiveStreamRequest(uint32_t flags,
+                                 uint32_t opaque,
+                                 uint16_t vbucket,
+                                 uint64_t start_seqno,
+                                 uint64_t end_seqno,
+                                 uint64_t vbucket_uuid,
+                                 uint64_t snap_start_seqno,
+                                 uint64_t snap_end_seqno);
 };
 
 using mock_dcp_producer_t = SingleThreadedRCPtr<MockDcpProducer>;

tests/mock/mock_stream.h

@@ -17,6 +17,8 @@
 
 #pragma once
 
+#include "dcp/consumer.h"
+#include "dcp/producer.h"
 #include "dcp/stream.h"
 
 /*
@@ -90,4 +92,13 @@ class MockActiveStream : public ActiveStream {
     void transitionStateToBackfilling() {
         transitionState(STREAM_BACKFILLING);
     }
+
+    virtual void getOutstandingItems(RCPtr<VBucket>& vb,
+                                     std::vector<queued_item>& items) override {
+        preGetOutstandingItemsCallback();
+        ActiveStream::getOutstandingItems(vb, items);
+    }
+
+    /// A callback to allow tests to inject code before we access the checkpoint
+    std::function<void()> preGetOutstandingItemsCallback = [] { return; };
 };

tests/module_tests/dcp_test.cc

@@ -103,7 +103,7 @@ class StreamTest : public DCPTest {
         vb0 = engine->getVBucket(0);
         EXPECT_TRUE(vb0) << "Failed to get valid VBucket object for id 0";
         EXPECT_FALSE(vb0->checkpointManager.registerCursor(
-                                                           producer->getName(),
+                                                           static_cast<ActiveStream*>(stream.get())->getCursorName(),
                                                            1, false,
                                                            MustSendCheckpointEnd::NO))
             << "Found an existing TAP cursor when attempting to register ours";
@@ -973,21 +973,8 @@ class ActiveStreamChkptProcessorTaskTest : public SingleThreadedEPStoreTest {
     }
 
     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()));
+        SingleThreadedEPStoreTest::notifyAndStepToCheckpoint(*producer,
+                                                              producers.get());
     }
 
     const void* cookie;