MB-29287: Merge branch 'couchbase/5.1.0' into 'couchbase/spock'

* couchbase/5.1.0:
  MB-29287: Test that takeover finds all items
  MB-29287: Give each ActiveStream a unique cursor name

Change-Id: I26962e40fac08dd6df229fdc8cd7be5654403036

Author: jimwwalker

engines/ep/CMakeLists.txt

@@ -242,6 +242,7 @@ TARGET_LINK_LIBRARIES(ep cJSON JSON_checker couchstore ${EP_FORESTDB_LIB}
 # 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

@@ -997,12 +997,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

@@ -104,10 +104,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

@@ -268,7 +268,8 @@ ActiveStream::ActiveStream(EventuallyPersistentEngine* e,
       chkptItemsExtractionInProgress(false),
       includeValue(includeVal),
       includeXattributes(includeXattrs),
-      filter(std::move(filter)) {
+      filter(std::move(filter)),
+      cursorName(n + std::to_string(cursorUID.fetch_add(1))) {
     const char* type = "";
     if (flags_ & DCP_ADD_STREAM_FLAG_TAKEOVER) {
         type = "takeover ";
@@ -364,7 +365,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
@@ -814,8 +815,9 @@ void ActiveStream::addTakeoverStats(ADD_STAT add_stat, const void *cookie,
                     add_stat, cookie);
 
     size_t vb_items = vb.getNumItems();
-    size_t chk_items = vb_items > 0 ?
-                vb.checkpointManager.getNumItemsForCursor(name_) : 0;
+    size_t chk_items =
+            vb_items > 0 ? vb.checkpointManager.getNumItemsForCursor(cursorName)
+                         : 0;
 
     size_t del_items = 0;
     try {
@@ -866,7 +868,8 @@ DcpResponse* ActiveStream::nextQueuedItem() {
 
 bool ActiveStream::nextCheckpointItem() {
     VBucketPtr 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;
@@ -957,7 +960,7 @@ void ActiveStream::getOutstandingItems(VBucketPtr &vb,
     chkptItemsExtractionInProgress.store(true);
 
     hrtime_t _begin_ = gethrtime();
-    vb->checkpointManager.getAllItemsForCursor(name_, items);
+    vb->checkpointManager.getAllItemsForCursor(cursorName, items);
     engine->getEpStats().dcpCursorsGetItemsHisto.add(
                                             (gethrtime() - _begin_) / 1000);
 
@@ -1195,7 +1198,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,
@@ -1273,11 +1277,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 "
@@ -1469,7 +1474,7 @@ void ActiveStream::transitionState(StreamState newState) {
             {
                 VBucketPtr vb = engine->getVBucket(vb_);
                 if (vb) {
-                    vb->checkpointManager.removeCursor(name_);
+                    vb->checkpointManager.removeCursor(cursorName);
                 }
                 break;
             }
@@ -1493,8 +1498,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 {
@@ -1534,7 +1539,7 @@ bool ActiveStream::dropCheckpointCursor_UNLOCKED() {
         }
     }
     /* Drop the existing cursor */
-    return vbucket->checkpointManager.removeCursor(name_);
+    return vbucket->checkpointManager.removeCursor(cursorName);
 }
 
 EXTENSION_LOG_LEVEL ActiveStream::getTransitionStateLogLevel(
@@ -1576,6 +1581,8 @@ bool ActiveStream::queueResponse(DcpResponse* resp) const {
     return true;
 }
 
+std::atomic<uint64_t> ActiveStream::cursorUID;
+
 NotifierStream::NotifierStream(EventuallyPersistentEngine* e,
                                dcp_producer_t p,
                                const std::string& name,

engines/ep/src/dcp/stream.h

@@ -299,9 +299,15 @@ class ActiveStream : public Stream {
         return currentSeparator;
     }
 
+    /// @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(VBucketPtr &vb, std::vector<queued_item> &items);
+    virtual void getOutstandingItems(VBucketPtr& 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.
@@ -490,6 +496,17 @@ class ActiveStream : public Stream {
      * The filter the stream will use to decide which keys should be transmitted
      */
     std::unique_ptr<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) {
+    stream_t stream = new MockActiveStream(
+            static_cast<EventuallyPersistentEngine*>(&engine_),
+            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

@@ -129,6 +129,18 @@ class MockDcpProducer: public DcpProducer {
     BackfillManager& getBFM() {
         return *backfillMgr;
     }
+
+    /**
+     * 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);
 };
 
 using mock_dcp_producer_t = SingleThreadedRCPtr<MockDcpProducer>;

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"
 
@@ -148,6 +150,15 @@ class MockActiveStream : public ActiveStream {
     void setState(StreamState state) {
         state_ = state;
     }
+
+    virtual void getOutstandingItems(VBucketPtr& 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; };
 };
 
 /**

engines/ep/tests/module_tests/dcp_test.cc

@@ -128,10 +128,12 @@ class StreamTest : public DCPTest,
                                       includeXattrs);
 
         EXPECT_FALSE(vb0->checkpointManager.registerCursor(
-                                                           producer->getName(),
-                                                           1, false,
-                                                           MustSendCheckpointEnd::NO))
-            << "Found an existing TAP cursor when attempting to register ours";
+                static_cast<ActiveStream*>(stream.get())->getCursorName(),
+                1,
+                false,
+                MustSendCheckpointEnd::NO))
+                << "Found an existing TAP cursor when attempting to register "
+                   "ours";
     }
 
     void destroy_dcp_stream() {
@@ -2291,21 +2293,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.get());
     }
 
     const void* cookie;