Address heap use after free issue in Dcp BackfillManager

21:23:11 WARNING: ThreadSanitizer: heap-use-after-free (pid=8561)
21:23:11   Read of size 8 at 0x7d240000d6a8 by thread T15:
21:23:11     #0 BackfillManager::backfill() /home/couchbase/jenkins/workspace/ep-engine-threadsanitizer-master/ep-engine/src/dcp/backfill-manager.cc:250 (ep.so+0x00000004f35a)
21:23:11     couchbase#1 BackfillManagerTask::run() /home/couchbase/jenkins/workspace/ep-engine-threadsanitizer-master/ep-engine/src/dcp/backfill-manager.cc:43 (ep.so+0x00000004ee6f)
21:23:11     couchbase#2 ExecutorThread::run() /home/couchbase/jenkins/workspace/ep-engine-threadsanitizer-master/ep-engine/src/executorthread.cc:115 (ep.so+0x0000000f1736)
21:23:11     couchbase#3 launch_executor_thread(void*) /home/couchbase/jenkins/workspace/ep-engine-threadsanitizer-master/ep-engine/src/executorthread.cc:33 (ep.so+0x0000000f12e5)
21:23:11     couchbase#4 platform_thread_wrap(void*) /home/couchbase/jenkins/workspace/ep-engine-threadsanitizer-master/platform/src/cb_pthreads.cc:54 (libplatform.so.0.1.0+0x00000000551b)
21:23:11
21:23:11   Previous write of size 8 at 0x7d240000d6a8 by thread T15:
21:23:11     #0 operator delete(void*) <null> (engine_testapp+0x00000046485b)
21:23:11     couchbase#1 DcpProducer::~DcpProducer() /home/couchbase/jenkins/workspace/ep-engine-threadsanitizer-master/ep-engine/src/dcp/producer.cc:167 (ep.so+0x00000006377b)
21:23:11     couchbase#2 DcpProducer::~DcpProducer() /home/couchbase/jenkins/workspace/ep-engine-threadsanitizer-master/ep-engine/src/dcp/producer.cc:165 (ep.so+0x000000063a45)
21:23:11     couchbase#3 ActiveStream::~ActiveStream() /home/couchbase/jenkins/workspace/ep-engine-threadsanitizer-master/ep-engine/src/atomic.h:272 (ep.so+0x00000006ed6d)
21:23:11     couchbase#4 ActiveStream::~ActiveStream() /home/couchbase/jenkins/workspace/ep-engine-threadsanitizer-master/ep-engine/src/dcp/stream.cc:200 (ep.so+0x00000006f8b5)
21:23:11     couchbase#5 BackfillManager::backfill() /home/couchbase/jenkins/workspace/ep-engine-threadsanitizer-master/ep-engine/src/atomic.h:272 (ep.so+0x00000004f345)
21:23:11     #6 BackfillManagerTask::run() /home/couchbase/jenkins/workspace/ep-engine-threadsanitizer-master/ep-engine/src/dcp/backfill-manager.cc:43 (ep.so+0x00000004ee6f)
21:23:11     #7 ExecutorThread::run() /home/couchbase/jenkins/workspace/ep-engine-threadsanitizer-master/ep-engine/src/executorthread.cc:115 (ep.so+0x0000000f1736)
21:23:11     #8 launch_executor_thread(void*) /home/couchbase/jenkins/workspace/ep-engine-threadsanitizer-master/ep-engine/src/executorthread.cc:33 (ep.so+0x0000000f12e5)
21:23:11     #9 platform_thread_wrap(void*) /home/couchbase/jenkins/workspace/ep-engine-threadsanitizer-master/platform/src/cb_pthreads.cc:54 (libplatform.so.0.1.0+0x00000000551b)

Change-Id: I3c63215791c23de49b5304654115fd4c558a3328
Reviewed-on: http://review.couchbase.org/58619
Tested-by: buildbot <build@couchbase.com>
Reviewed-by: Chiyoung Seo <chiyoung@couchbase.com>

Author: abhinavdangeti

src/connmap.cc

@@ -1214,7 +1214,7 @@ void DcpConnMap::notifyBackfillManagerTasks() {
     for (; itr != map_.end(); ++itr) {
         DcpProducer* producer = dynamic_cast<DcpProducer*> (itr->second.get());
         if (producer) {
-            producer->getBackfillManager()->wakeUpTask();
+            producer->notifyBackfillManager();
         }
     }
 }

src/dcp/backfill-manager.cc

@@ -26,17 +26,17 @@ static const size_t sleepTime = 1;
 
 class BackfillManagerTask : public GlobalTask {
 public:
-    BackfillManagerTask(EventuallyPersistentEngine* e, BackfillManager* mgr,
+    BackfillManagerTask(EventuallyPersistentEngine* e, backfill_manager_t mgr,
                         const Priority &p, double sleeptime = 0,
-                        bool shutdown = false)
-        : GlobalTask(e, p, sleeptime, shutdown), manager(mgr) {}
+                        bool completeBeforeShutdown = false)
+        : GlobalTask(e, p, sleeptime, completeBeforeShutdown), manager(mgr) {}
 
     bool run();
 
     std::string getDescription();
 
 private:
-    BackfillManager* manager;
+    backfill_manager_t manager;
 };
 
 bool BackfillManagerTask::run() {
@@ -88,10 +88,6 @@ void BackfillManager::addStats(connection_t conn, ADD_STAT add_stat,
 }
 
 BackfillManager::~BackfillManager() {
-    if (managerTask) {
-        managerTask->cancel();
-    }
-
     while (!activeBackfills.empty()) {
         DCPBackfill* backfill = activeBackfills.front();
         activeBackfills.pop_front();
@@ -116,6 +112,15 @@ BackfillManager::~BackfillManager() {
     }
 }
 
+void BackfillManager::terminate() {
+    LockHolder lh(lock);
+    if (managerTask) {
+        managerTask->cancel();
+        managerTask.reset();
+    }
+}
+
+
 void BackfillManager::schedule(stream_t stream, uint64_t start, uint64_t end) {
     LockHolder lh(lock);
     if (engine->getDcpConnMap().canAddBackfillToActiveQ()) {

src/dcp/backfill-manager.h

@@ -27,12 +27,14 @@
 
 class EventuallyPersistentEngine;
 
-class BackfillManager {
+class BackfillManager : public RCValue {
 public:
     BackfillManager(EventuallyPersistentEngine* e);
 
     ~BackfillManager();
 
+    void terminate();
+
     void addStats(connection_t conn, ADD_STAT add_stat, const void *c);
 
     void schedule(stream_t stream, uint64_t start, uint64_t end);

src/dcp/dcp-types.h

@@ -40,3 +40,7 @@ typedef SingleThreadedRCPtr<Stream> stream_t;
 // Implementation defined in dcp/stream.h
 class PassiveStream;
 typedef RCPtr<PassiveStream> passive_stream_t;
+
+// Implementation defined in dcp/backfill-manager.h
+class BackfillManager;
+typedef RCPtr<BackfillManager> backfill_manager_t;

src/dcp/producer.cc

@@ -163,8 +163,8 @@ DcpProducer::DcpProducer(EventuallyPersistentEngine &e, const void *cookie,
 }
 
 DcpProducer::~DcpProducer() {
+    backfillMgr->terminate();
     delete rejectResp;
-    delete backfillMgr;
 }
 
 ENGINE_ERROR_CODE DcpProducer::streamRequest(uint32_t flags,
@@ -652,6 +652,23 @@ ENGINE_ERROR_CODE DcpProducer::closeStream(uint32_t opaque, uint16_t vbucket) {
     return ret;
 }
 
+void DcpProducer::notifyBackfillManager() {
+    backfillMgr->wakeUpTask();
+}
+
+bool DcpProducer::recordBackfillManagerBytesRead(uint32_t bytes) {
+    return backfillMgr->bytesRead(bytes);
+}
+
+void DcpProducer::recordBackfillManagerBytesSent(uint32_t bytes) {
+    backfillMgr->bytesSent(bytes);
+}
+
+void DcpProducer::scheduleBackfillManager(stream_t s,
+                                          uint64_t start, uint64_t end) {
+    backfillMgr->schedule(s, start, end);
+}
+
 void DcpProducer::addStats(ADD_STAT add_stat, const void *c) {
     Producer::addStats(add_stat, c);
 
@@ -671,9 +688,7 @@ void DcpProducer::addStats(ADD_STAT add_stat, const void *c) {
             supportsCursorDropping ? "ELIGIBLE" : "NOT_ELIGIBLE",
             add_stat, c);
 
-    if (backfillMgr) {
-        backfillMgr->addStats(this, add_stat, c);
-    }
+    backfillMgr->addStats(this, add_stat, c);
 
     log.addStats(add_stat, c);
 

src/dcp/producer.h

@@ -107,9 +107,10 @@ class DcpProducer : public Producer {
 
     void notifyStreamReady(uint16_t vbucket, bool schedule);
 
-    BackfillManager* getBackfillManager() {
-        return backfillMgr;
-    }
+    void notifyBackfillManager();
+    bool recordBackfillManagerBytesRead(uint32_t bytes);
+    void recordBackfillManagerBytesSent(uint32_t bytes);
+    void scheduleBackfillManager(stream_t s, uint64_t start, uint64_t end);
 
     bool isExtMetaDataEnabled () {
         return enableExtMetaData;
@@ -230,7 +231,7 @@ class DcpProducer : public Producer {
     Couchbase::RelaxedAtomic<rel_time_t> lastSendTime;
     BufferLog log;
 
-    BackfillManager* backfillMgr;
+    backfill_manager_t backfillMgr;
 
     // Guards all accesses to streams map. If only reading elements in streams
     // (i.e. not adding / removing elements) then can acquire ReadLock, even

src/dcp/stream.cc

@@ -314,7 +314,7 @@ bool ActiveStream::backfillReceived(Item* itm, backfill_source_t backfill_source
     }
     LockHolder lh(streamMutex);
     if (state_ == STREAM_BACKFILLING) {
-        if (!producer->getBackfillManager()->bytesRead(itm->size())) {
+        if (!producer->recordBackfillManagerBytesRead(itm->size())) {
             delete itm;
             return false;
         }
@@ -425,7 +425,7 @@ DcpResponse* ActiveStream::backfillPhase() {
          resp->getEvent() == DCP_DELETION ||
          resp->getEvent() == DCP_EXPIRATION)) {
         MutationResponse* m = static_cast<MutationResponse*>(resp);
-        producer->getBackfillManager()->bytesSent(m->getItem()->size());
+        producer->recordBackfillManagerBytesSent(m->getItem()->size());
         bufferedBackfill.bytes.fetch_sub(m->getItem()->size());
         bufferedBackfill.items--;
         if (backfillRemaining.load(std::memory_order_relaxed) > 0) {
@@ -740,7 +740,7 @@ void ActiveStream::endStream(end_stream_status_t reason) {
             // If Stream were in Backfilling state, clear out the
             // backfilled items to clear up the backfill buffer.
             clear_UNLOCKED();
-            producer->getBackfillManager()->bytesSent(bufferedBackfill.bytes);
+            producer->recordBackfillManagerBytesSent(bufferedBackfill.bytes);
             bufferedBackfill.bytes = 0;
             bufferedBackfill.items = 0;
         }
@@ -801,8 +801,7 @@ void ActiveStream::scheduleBackfill() {
         bool tryBackfill = isFirstItem || flags_ & DCP_ADD_STREAM_FLAG_DISKONLY;
 
         if (backfillStart <= backfillEnd && tryBackfill) {
-            BackfillManager* backfillMgr = producer->getBackfillManager();
-            backfillMgr->schedule(this, backfillStart, backfillEnd);
+            producer->scheduleBackfillManager(this, backfillStart, backfillEnd);
             isBackfillTaskRunning.store(true);
         } else {
             if (flags_ & DCP_ADD_STREAM_FLAG_DISKONLY) {