Address data race in DcpConsumer, by acquiring readyMutex

WARNING: ThreadSanitizer: data race (pid=27652)

  Write of size 8 at 0x7d08000443c0 by main thread (mutexes: write M57876):
    #0 operator delete(void*) <null>:0 (engine_testapp+0x000000050e7b)
    couchbase#1 __gnu_cxx::new_allocator<std::_List_node<unsigned short> >::deallocate(std::_List_node<unsigned short>*, unsigned long) /usr/bin/../lib/gcc/x86_64-linux-gnu/4.8/../../../../include/c++/4.8/ext/new_allocator.h:110 (ep.so+0x00000005d69a)
    couchbase#2 DcpConsumer::step(dcp_message_producers*) /home/abhinav/couchbase/ep-engine/src/dcp/consumer.cc:516 (ep.so+0x00000005c5cc)
    couchbase#3 EvpDcpStep(engine_interface*, void const*, dcp_message_producers*) /home/abhinav/couchbase/ep-engine/src/ep_engine.cc:1479 (ep.so+0x0000000b480b)
    couchbase#4 mock_dcp_step(engine_interface*, void const*, dcp_message_producers*) /home/abhinav/couchbase/memcached/programs/engine_testapp/engine_testapp.cc:476 (engine_testapp+0x0000000bb055)
    couchbase#5 dcp_step(engine_interface*, engine_interface_v1*, void const*) /home/abhinav/couchbase/ep-engine/tests/ep_test_apis.cc:1219 (ep_testsuite.so+0x0000000b61bd)
    #6 test_chk_manager_rollback(engine_interface*, engine_interface_v1*) /home/abhinav/couchbase/ep-engine/tests/ep_testsuite.cc:5526 (ep_testsuite.so+0x0000000809b4)
    #7 execute_test(test, char const*, char const*) /home/abhinav/couchbase/memcached/programs/engine_testapp/engine_testapp.cc:1090 (engine_testapp+0x0000000b952c)
    #8 __libc_start_main /build/buildd/eglibc-2.19/csu/libc-start.c:287 (libc.so.6+0x000000021ec4)

  Previous write of size 8 at 0x7d08000443c0 by thread T16:
    #0 operator new(unsigned long) <null>:0 (engine_testapp+0x00000005090d)
    couchbase#1 __gnu_cxx::new_allocator<std::_List_node<unsigned short> >::allocate(unsigned long, void const*) /usr/bin/../lib/gcc/x86_64-linux-gnu/4.8/../../../../include/c++/4.8/ext/new_allocator.h:104 (ep.so+0x00000005f265)
    couchbase#2 PassiveStream::reconnectStream(RCPtr<VBucket>&, unsigned int, unsigned long) /home/abhinav/couchbase/ep-engine/src/dcp/stream.cc:1104 (ep.so+0x000000076f5f)
    couchbase#3 DcpConsumer::doRollback(unsigned int, unsigned short, unsigned long) /home/abhinav/couchbase/ep-engine/src/dcp/consumer.cc:676 (ep.so+0x00000005db67)
    couchbase#4 RollbackTask::run() /home/abhinav/couchbase/ep-engine/src/dcp/consumer.cc:574 (ep.so+0x00000005d9d4)
    couchbase#5 ExecutorThread::run() /home/abhinav/couchbase/ep-engine/src/executorthread.cc:115 (ep.so+0x0000000f834c)
    #6 launch_executor_thread(void*) /home/abhinav/couchbase/ep-engine/src/executorthread.cc:33 (ep.so+0x0000000f7eb5)
    #7 platform_thread_wrap /home/abhinav/couchbase/platform/src/cb_pthreads.c:23 (libplatform.so.0.1.0+0x000000003d71)

Change-Id: I196a78e54bf8014967a51cdb081126597153f77b
Reviewed-on: http://review.couchbase.org/55881
Tested-by: buildbot <[email protected]>
Reviewed-by: Chiyoung Seo <[email protected]>

Author: abhinavdangeti

src/dcp/consumer.cc

@@ -802,13 +802,15 @@ DcpResponse* DcpConsumer::getNextItem() {
 }
 
 void DcpConsumer::notifyStreamReady(uint16_t vbucket) {
+    LockHolder lh(readyMutex);
     std::list<uint16_t>::iterator iter =
         std::find(ready.begin(), ready.end(), vbucket);
     if (iter != ready.end()) {
         return;
     }
 
     ready.push_back(vbucket);
+    lh.unlock();
 
     engine_.getDcpConnMap().notifyPausedConnection(this, true);
 }