MB-26979: ChkptProcessorTask should not own dcp stream objs

Currently the ActiveStreamCheckpointProcessorTask co-owns the
corresponding stream object. So if the producer connection
(co-owner) quickly closes the stream and opens a new stream on the same
vbucket (that is the streams map of the producer will contain
new stream object), then the ActiveStreamCheckpointProcessorTask
will contain any entry to the older stream. This could result in a
case where the processor task is not run for the new stream which
could result in a DCP hang.

This commit fixes the issue by making sure that the processor
task only contains the vbucket id and the stream is looked up from
the producer streams map when needed. However this requires the
ActiveStreamCheckpointProcessorTask to hold a shared reference to
the producer though the producer obj holds a shared reference to
the task (thereby resulting in a cyclic reference). Hence in the delete
path, the cyclic reference is broken by manually deleting the producer
reference.

The problem of cyclic reference can be averted in the master branch
by the use of shared/weak ptr.

Also, to test the code, some refactoring of test code is done.

Change-Id: I4b16bb81aac6f45a137affa9870be6f1416e9464
Reviewed-on: http://review.couchbase.org/86277
Well-Formed: Build Bot <[email protected]>
Tested-by: Build Bot <[email protected]>
Reviewed-by: Jim Walker <[email protected]>

Author: manudhundi

src/connmap.cc

@@ -1146,7 +1146,7 @@ void DcpConnMap::closeStreams(CookieToConnectionMap& map) {
         DcpProducer* producer = dynamic_cast<DcpProducer*> (itr.second.get());
         if (producer) {
             producer->closeAllStreams();
-            producer->clearCheckpointProcessorTaskQueues();
+            producer->cancelCheckpointCreatorTask();
             // The producer may be in EWOULDBLOCK (if it's idle), therefore
             // notify him to ensure the front-end connection can close the TCP
             // connection.
@@ -1207,7 +1207,7 @@ void DcpConnMap::disconnect(const void *cookie) {
         DcpProducer* producer = dynamic_cast<DcpProducer*> (conn.get());
         if (producer) {
             producer->closeAllStreams();
-            producer->clearCheckpointProcessorTaskQueues();
+            producer->cancelCheckpointCreatorTask();
         } else {
             // Cancel consumer's processer task before closing all streams
             static_cast<DcpConsumer*>(conn.get())->cancelTask();

src/dcp/producer.cc

@@ -184,14 +184,18 @@ DcpProducer::DcpProducer(EventuallyPersistentEngine &e, const void *cookie,
 
     backfillMgr.reset(new BackfillManager(&engine_));
 
-    checkpointCreatorTask = new ActiveStreamCheckpointProcessorTask(e);
+    checkpointCreatorTask = new ActiveStreamCheckpointProcessorTask(e, this);
     ExecutorPool::get()->schedule(checkpointCreatorTask, AUXIO_TASK_IDX);
 }
 
 DcpProducer::~DcpProducer() {
     backfillMgr.reset();
     delete rejectResp;
+}
 
+void DcpProducer::cancelCheckpointCreatorTask() {
+    static_cast<ActiveStreamCheckpointProcessorTask*>(
+            checkpointCreatorTask.get())->cancelTask();
     ExecutorPool::get()->cancel(checkpointCreatorTask->getId());
 }
 
@@ -1037,8 +1041,3 @@ void DcpProducer::scheduleCheckpointProcessorTask(stream_t s) {
     static_cast<ActiveStreamCheckpointProcessorTask*>(checkpointCreatorTask.get())
         ->schedule(s);
 }
-
-void DcpProducer::clearCheckpointProcessorTaskQueues() {
-    static_cast<ActiveStreamCheckpointProcessorTask*>(checkpointCreatorTask.get())
-        ->clearQueues();
-}

src/dcp/producer.h

@@ -32,7 +32,13 @@ class DcpProducer : public Producer {
     DcpProducer(EventuallyPersistentEngine &e, const void *cookie,
                 const std::string &n, bool notifyOnly);
 
-    ~DcpProducer();
+    virtual ~DcpProducer();
+
+    /**
+     * Clears active stream checkpoint processor task's queue, resets its
+     * shared reference to the producer and cancels the task.
+     */
+    void cancelCheckpointCreatorTask();
 
     ENGINE_ERROR_CODE streamRequest(uint32_t flags, uint32_t opaque,
                                     uint16_t vbucket, uint64_t start_seqno,
@@ -201,10 +207,14 @@ class DcpProducer : public Producer {
     */
     void scheduleCheckpointProcessorTask(stream_t s);
 
-    /*
-        Clears active stream checkpoint processor task's queue.
-    */
-    void clearCheckpointProcessorTaskQueues();
+    /**
+     * Returns a shared reference to the stream associated with the vbucket
+     *
+     * @param vbid Vbucket id
+     *
+     * @return shared reference ptr to the stream associated
+     */
+    stream_t findStreamByVbid(uint16_t vbid);
 
 protected:
 
@@ -218,13 +228,14 @@ class DcpProducer : public Producer {
         Couchbase::RelaxedAtomic<bool> enabled;
     } noopCtx;
 
+    ExTask checkpointCreatorTask;
+
 private:
 
 
     DcpResponse* getNextItem();
 
     size_t getItemsRemaining();
-    stream_t findStreamByVbid(uint16_t vbid);
 
     std::string priority;
 
@@ -256,9 +267,7 @@ class DcpProducer : public Producer {
     AtomicValue<size_t> itemsSent;
     AtomicValue<size_t> totalBytesSent;
 
-    ExTask checkpointCreatorTask;
     static const uint32_t defaultNoopInerval;
-
 };
 
 #endif  // SRC_DCP_PRODUCER_H_

src/dcp/stream.cc

@@ -798,20 +798,23 @@ void ActiveStreamCheckpointProcessorTask::wakeup() {
 }
 
 void ActiveStreamCheckpointProcessorTask::schedule(stream_t stream) {
-    pushUnique(stream);
+    pushUnique(stream->getVBucket());
 
     bool expected = false;
     if (notified.compare_exchange_strong(expected, true)) {
         wakeup();
     }
 }
 
-void ActiveStreamCheckpointProcessorTask::clearQueues() {
+void ActiveStreamCheckpointProcessorTask::cancelTask() {
     LockHolder lh(workQueueLock);
     while (!queue.empty()) {
         queue.pop();
     }
     queuedVbuckets.clear();
+    /* Reset the producer while holding the lock as it is a
+       SingleThreadedRCPtr */
+    producer.reset();
 }
 
 void ActiveStream::nextCheckpointItemTask() {

src/dcp/stream.h

@@ -386,12 +386,14 @@ class ActiveStream : public Stream {
 
 class ActiveStreamCheckpointProcessorTask : public GlobalTask {
 public:
-    ActiveStreamCheckpointProcessorTask(EventuallyPersistentEngine& e)
+    ActiveStreamCheckpointProcessorTask(EventuallyPersistentEngine& e,
+                                        dcp_producer_t p)
         : GlobalTask(&e, TaskId::ActiveStreamCheckpointProcessorTask,
                      INT_MAX, false),
       notified(false),
       iterationsBeforeYield(e.getConfiguration()
-                            .getDcpProducerSnapshotMarkerYieldLimit()) { }
+                            .getDcpProducerSnapshotMarkerYieldLimit()),
+      producer(p) { }
 
     std::string getDescription() {
         std::string rv("Process checkpoint(s) for DCP producer");
@@ -401,17 +403,38 @@ class ActiveStreamCheckpointProcessorTask : public GlobalTask {
     bool run();
     void schedule(stream_t stream);
     void wakeup();
-    void clearQueues();
+
+    /* Clears the queues and resets the producer reference */
+    void cancelTask();
+
+    /* Returns the number of unique streams waiting to be processed */
+    size_t queueSize() {
+        LockHolder lh(workQueueLock);
+        return queue.size();
+    }
 
 private:
 
     stream_t queuePop() {
         stream_t rval;
-        LockHolder lh(workQueueLock);
-        if (!queue.empty()) {
-            rval = queue.front();
+        uint16_t vbid = 0;
+        dcp_producer_t producerRefCpy;
+        {
+            LockHolder lh(workQueueLock);
+            if (queue.empty()) {
+                return rval;
+            }
+            vbid = queue.front();
             queue.pop();
-            queuedVbuckets.erase(rval->getVBucket());
+            queuedVbuckets.erase(vbid);
+            /* Get a copy of refPtr as we are releasing the workQueueLock */
+            producerRefCpy = producer;
+        }
+
+        /* findStreamByVbid acquires DcpProducer::streamsMutex, hence called
+           without acquiring workQueueLock */
+        if (producerRefCpy) {
+            return producerRefCpy->findStreamByVbid(vbid);
         }
         return rval;
     }
@@ -421,25 +444,34 @@ class ActiveStreamCheckpointProcessorTask : public GlobalTask {
         return queue.empty();
     }
 
-    void pushUnique(stream_t stream) {
+    void pushUnique(uint16_t vbid) {
         LockHolder lh(workQueueLock);
-        if (queuedVbuckets.count(stream->getVBucket()) == 0) {
-            queue.push(stream);
-            queuedVbuckets.insert(stream->getVBucket());
+        if (queuedVbuckets.count(vbid) == 0) {
+            queue.push(vbid);
+            queuedVbuckets.insert(vbid);
         }
     }
 
     Mutex workQueueLock;
 
     /**
-     * Maintain a queue of unique stream_t
+     * Maintain a queue of unique vbucket ids for which stream should be
+     * processed.
      * There's no need to have the same stream in the queue more than once
+     *
+     * The streams are kept in the 'streams map' of the producer object. We
+     * should not hold a shared reference to the stream object here in order to
+     * avoid multiple stream ownership issues
      */
-    std::queue<stream_t> queue;
-    std::set<uint16_t> queuedVbuckets;
+    std::queue<uint16_t> queue;
+    std::unordered_set<uint16_t> queuedVbuckets;
 
     AtomicValue<bool> notified;
     size_t iterationsBeforeYield;
+
+    /* shared reference to the producer object, should be deleted when the task
+       is stopped because the producer object contains a reference to this */
+    dcp_producer_t producer;
 };
 
 class NotifierStream : public Stream {

tests/mock/mock_dcp_producer.h

@@ -18,6 +18,7 @@
 #pragma once
 
 #include "dcp/producer.h"
+#include "dcp/stream.h"
 
 /*
  * Mock of the DcpProducer class.  Wraps the real DcpProducer, but exposes
@@ -54,4 +55,11 @@ class MockDcpProducer: public DcpProducer {
     bool getNoopEnabled() {
         return noopCtx.enabled;
     }
-};
+
+    ActiveStreamCheckpointProcessorTask& getCheckpointSnapshotTask() const {
+        return *static_cast<ActiveStreamCheckpointProcessorTask*>(
+                        checkpointCreatorTask.get());
+    }
+};
+
+using mock_dcp_producer_t = SingleThreadedRCPtr<MockDcpProducer>;