MB-29287: Give each ActiveStream a unique cursor name

To ensure that when the background snapshot processor task runs
and we have closed/created the ActiveStream, if the task gets a handle
on the closed stream we must be sure it does not obtain items destined
for the new stream. Previously with each ActiveStream just using the
name of its producer, the closed stream was able to drain items which
the new stream needed.

Adding a monotonic atomic to the ActiveStream class and appending a
new value to the name we use for the cursor ensures concurrent streams
cannot interfere with each other.

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

Author: jimwwalker

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();
 

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,6 +257,11 @@ 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);
@@ -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;
+
 };
 
 

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";