MB-19404: [BP] Address data race in DCP-Producer seen while making a stats request

WARNING: ThreadSanitizer: data race (pid=82258)
  Read of size 1 at 0x7d4c0000a208 by thread T11 (mutexes: write M2483, write M19044):
    #0 DcpProducer::addStats(void (*)(char const*, unsigned short, char const*, unsigned int, void const*), void const*) /home/couchbase/jenkins/workspace/ep-engine-threadsanitizer-master/ep-engine/src/dcp/producer.cc:601 (ep.so+0x000000063e2d)
    #1 ConnStatBuilder::operator()(SingleThreadedRCPtr<ConnHandler>&) /home/couchbase/jenkins/workspace/ep-engine-threadsanitizer-master/ep-engine/src/ep_engine.cc:3903 (ep.so+0x0000000d6931)
    #2 EventuallyPersistentEngine::doDcpStats(void const*, void (*)(char const*, unsigned short, char const*, unsigned int, void const*)) /home/couchbase/jenkins/workspace/ep-engine-threadsanitizer-master/ep-engine/src/ep_engine.cc:4160 (ep.so+0x0000000b904a)
    #3 EventuallyPersistentEngine::getStats(void const*, char const*, int, void (*)(char const*, unsigned short, char const*, unsigned int, void const*)) /home/couchbase/jenkins/workspace/ep-engine-threadsanitizer-master/ep-engine/src/ep_engine.cc:4580 (ep.so+0x0000000bcba4)
    #4 EventuallyPersistentStore::snapshotStats() /home/couchbase/jenkins/workspace/ep-engine-threadsanitizer-master/ep-engine/src/ep.cc:1700 (ep.so+0x000000088386)
    #5 StatSnap::run() /home/couchbase/jenkins/workspace/ep-engine-threadsanitizer-master/ep-engine/src/tasks.cc:98 (ep.so+0x00000012ba26)
    #6 ExecutorThread::run() /home/couchbase/jenkins/workspace/ep-engine-threadsanitizer-master/ep-engine/src/executorthread.cc:115 (ep.so+0x0000000eaeed)
    #7 launch_executor_thread(void*) /home/couchbase/jenkins/workspace/ep-engine-threadsanitizer-master/ep-engine/src/executorthread.cc:33 (ep.so+0x0000000eaab5)
    #8 platform_thread_wrap(void*) /home/couchbase/jenkins/workspace/ep-engine-threadsanitizer-master/platform/src/cb_pthreads.cc:53 (libplatform.so.0.1.0+0x0000000048bb)

  Previous write of size 1 at 0x7d4c0000a208 by main thread:
    #0 DcpProducer::handleResponse(protocol_binary_response_header*) /home/couchbase/jenkins/workspace/ep-engine-threadsanitizer-master/ep-engine/src/dcp/producer.cc:547 (ep.so+0x000000063231)
    #1 EvpDcpResponseHandler(engine_interface*, void const*, protocol_binary_response_header*) /home/couchbase/jenkins/workspace/ep-engine-threadsanitizer-master/ep-engine/src/ep_engine.cc:1765 (ep.so+0x0000000ae08b)
    #2 mock_dcp_response_handler(engine_interface*, void const*, protocol_binary_response_header*) /home/couchbase/jenkins/workspace/ep-engine-threadsanitizer-master/memcached/programs/engine_testapp/engine_testapp.cc:796 (engine_testapp+0x0000004c68e5)
    #3 sendDcpAck(engine_interface*, engine_interface_v1*, void const*, protocol_binary_command, protocol_binary_response_status, unsigned int) /home/couchbase/jenkins/workspace/ep-engine-threadsanitizer-master/ep-engine/tests/ep_test_apis.cc:983 (ep_testsuite.so+0x0000000a6a22)
    #4 test_dcp_noop(engine_interface*, engine_interface_v1*) /home/couchbase/jenkins/workspace/ep-engine-threadsanitizer-master/ep-engine/tests/ep_testsuite.cc:3975 (ep_testsuite.so+0x000000068deb)
    #5 execute_test(test, char const*, char const*) /home/couchbase/jenkins/workspace/ep-engine-threadsanitizer-master/memcached/programs/engine_testapp/engine_testapp.cc:1090 (engine_testapp+0x0000004c4192)
    #6 __libc_start_main /build/buildd/eglibc-2.15/csu/libc-start.c:226 (libc.so.6+0x00000002176c)

(Reviewed-on: http://review.couchbase.org/56306)

Change-Id: Ice7236da5cc885d9e7612894ba3d37e357e13b4a
Reviewed-on: http://review.couchbase.org/63440
Well-Formed: buildbot <build@couchbase.com>
Reviewed-by: Dave Rigby <daver@couchbase.com>
Tested-by: buildbot <build@couchbase.com>
Reviewed-by: Chiyoung Seo <chiyoung@couchbase.com>

Author: abhinavdangeti

src/dcp-producer.h

@@ -23,6 +23,8 @@
 #include "tapconnection.h"
 #include "dcp-stream.h"
 
+#include <relaxed_atomic.h>
+
 class DcpResponse;
 
 class DcpProducer : public Producer {
@@ -283,8 +285,8 @@ class DcpProducer : public Producer {
         rel_time_t sendTime;
         uint32_t opaque;
         uint32_t noopInterval;
-        bool pendingRecv;
-        bool enabled;
+        Couchbase::RelaxedAtomic<bool> pendingRecv;
+        Couchbase::RelaxedAtomic<bool> enabled;
     } noopCtx;
 
     DcpResponse *rejectResp; // stash response for retry if E2BIG was hit

tests/ep_testsuite.cc

@@ -3314,15 +3314,14 @@ static enum test_result test_dcp_noop(ENGINE_HANDLE *h, ENGINE_HANDLE_V1 *h1) {
         if (err == ENGINE_DISCONNECT) {
             done = true;
         } else {
-            std::string stat;
             if (dcp_last_op == PROTOCOL_BINARY_CMD_DCP_NOOP) {
                 done = true;
-                stat = get_str_stat(h, h1, "eq_dcpq:unittest:noop_wait", "dcp");
-                check(stat.compare("true") == 0, "Didn't send noop");
+                checkeq(1, get_int_stat(h, h1, "eq_dcpq:unittest:noop_wait", "dcp"),
+                        "Didn't send noop");
                 sendDcpAck(h, h1, cookie, PROTOCOL_BINARY_CMD_DCP_NOOP,
                            PROTOCOL_BINARY_RESPONSE_SUCCESS, dcp_last_opaque);
-                stat = get_str_stat(h, h1, "eq_dcpq:unittest:noop_wait", "dcp");
-                check(stat.compare("false") == 0, "Didn't ack noop");
+                checkeq(0, get_int_stat(h, h1, "eq_dcpq:unittest:noop_wait", "dcp"),
+                        "Didn't send noop");
             } else if (dcp_last_op != 0) {
                 abort();
             }
@@ -3365,10 +3364,9 @@ static enum test_result test_dcp_noop_fail(ENGINE_HANDLE *h,
             done = true;
             disconnected = true;
         } else {
-            std::string stat;
             if (dcp_last_op == PROTOCOL_BINARY_CMD_DCP_NOOP) {
-                stat = get_str_stat(h, h1, "eq_dcpq:unittest:noop_wait", "dcp");
-                check(stat.compare("true") == 0, "Didn't send noop");
+                checkeq(1, get_int_stat(h, h1, "eq_dcpq:unittest:noop_wait", "dcp"),
+                        "Didn't send noop");
                 testHarness.time_travel(201);
             } else if (dcp_last_op != 0) {
                 abort();