Address possible data races in PassiveStream context

WARNING: ThreadSanitizer: data race (pid=3212)

  Write of size 8 at 0x7d5000016908 by thread T5 (mutexes: write M26478):
    #0 PassiveStream::reconnectStream(RCPtr<VBucket>&, unsigned int, unsigned long) /home/abhinav/couchbase/ep-engine/src/dcp/stream.cc:1097 (ep.so+0x000000076c0f)
    couchbase#1 DcpConsumer::doRollback(unsigned int, unsigned short, unsigned long) /home/abhinav/couchbase/ep-engine/src/dcp/consumer.cc:676 (ep.so+0x00000005db67)
    couchbase#2 RollbackTask::run() /home/abhinav/couchbase/ep-engine/src/dcp/consumer.cc:574 (ep.so+0x00000005d9d4)
    couchbase#3 ExecutorThread::run() /home/abhinav/couchbase/ep-engine/src/executorthread.cc:112 (ep.so+0x0000000f8916)
    couchbase#4 launch_executor_thread(void*) /home/abhinav/couchbase/ep-engine/src/executorthread.cc:33 (ep.so+0x0000000f84b5)
    couchbase#5 platform_thread_wrap /home/abhinav/couchbase/platform/src/cb_pthreads.c:23 (libplatform.so.0.1.0+0x000000003d31)

  Previous read of size 8 at 0x7d5000016908 by main thread (mutexes: write M1367):
    #0 PassiveStream::setDead_UNLOCKED(end_stream_status_t, LockHolder*) /home/abhinav/couchbase/ep-engine/src/dcp/stream.cc:1046 (ep.so+0x0000000759ca)
    couchbase#1 PassiveStream::setDead(end_stream_status_t) /home/abhinav/couchbase/ep-engine/src/dcp/stream.cc:1056 (ep.so+0x0000000766d7)
    couchbase#2 DcpConsumer::closeAllStreams() /home/abhinav/couchbase/ep-engine/src/dcp/consumer.cc:860 (ep.so+0x00000005a006)
    couchbase#3 DcpConnMap::disconnect_UNLOCKED(void const*) /home/abhinav/couchbase/ep-engine/src/connmap.cc:1137 (ep.so+0x000000049972)
    couchbase#4 DcpConnMap::disconnect(void const*) /home/abhinav/couchbase/ep-engine/src/connmap.cc:1111 (ep.so+0x00000004969b)
    couchbase#5 EventuallyPersistentEngine::handleDisconnect(void const*) /home/abhinav/couchbase/ep-engine/src/ep_engine.cc:6224 (ep.so+0x0000000d3bea)
    #6 EvpHandleDisconnect(void const*, ENGINE_EVENT_TYPE, void const*, void const*) /home/abhinav/couchbase/ep-engine/src/ep_engine.cc:1783 (ep.so+0x0000000b7046)
    #7 mock_perform_callbacks /home/abhinav/couchbase/memcached/programs/engine_testapp/mock_server.c:296 (engine_testapp+0x0000000bd420)
    #8 test_rollback_to_zero(engine_interface*, engine_interface_v1*) /home/abhinav/couchbase/ep-engine/tests/ep_testsuite.cc:5434 (ep_testsuite.so+0x00000007f45f)
    #9 execute_test(test, char const*, char const*) /home/abhinav/couchbase/memcached/programs/engine_testapp/engine_testapp.cc:1090 (engine_testapp+0x0000000b946c)
    #10 __libc_start_main /build/buildd/eglibc-2.19/csu/libc-start.c:287 (libc.so.6+0x000000021ec4)

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

Author: abhinavdangeti

src/dcp/stream.cc

@@ -1046,7 +1046,7 @@ uint32_t PassiveStream::setDead_UNLOCKED(end_stream_status_t status,
     LOG(logLevel, "%s (vb %" PRId16 ") Setting stream to dead"
         " state, last_seqno is %" PRIu64 ", totalBytes is %" PRIu32 ","
         " droppedBytes is %" PRIu32 ", status is %s",
-        consumer->logHeader(), vb_, last_seqno, totalBytes,
+        consumer->logHeader(), vb_, last_seqno.load(), totalBytes,
         unackedBytes, getEndStreamStatusStr(status));
     return totalBytes;
 }
@@ -1094,7 +1094,7 @@ void PassiveStream::reconnectStream(RCPtr<VBucket> &vb,
         start_seqno, end_seqno_, snap_start_seqno_, snap_end_seqno_);
 
     LockHolder lh(streamMutex);
-    last_seqno = start_seqno;
+    last_seqno.store(start_seqno);
     pushToReadyQ(new StreamRequest(vb_, new_opaque, flags_, start_seqno,
                                   end_seqno_, vb_uuid_, snap_start_seqno_,
                                   snap_end_seqno_));
@@ -1124,30 +1124,30 @@ ENGINE_ERROR_CODE PassiveStream::messageReceived(DcpResponse* resp) {
         {
             MutationResponse* m = static_cast<MutationResponse*>(resp);
             uint64_t bySeqno = m->getBySeqno();
-            if (bySeqno <= last_seqno) {
+            if (bySeqno <= last_seqno.load()) {
                 LOG(EXTENSION_LOG_WARNING, "%s (vb %d) Erroneous (out of "
                     "sequence) mutation received, with opaque: %" PRIu32 ", its "
                     "seqno (%" PRIu64 ") is not greater than last received seqno "
                     "(%" PRIu64 "); Dropping mutation!", consumer->logHeader(),
-                    vb_, opaque_, bySeqno, last_seqno);
+                    vb_, opaque_, bySeqno, last_seqno.load());
                 delete m;
                 return ENGINE_ERANGE;
             }
-            last_seqno = bySeqno;
+            last_seqno.store(bySeqno);
             break;
         }
         case DCP_SNAPSHOT_MARKER:
         {
             SnapshotMarker* s = static_cast<SnapshotMarker*>(resp);
             uint64_t snapStart = s->getStartSeqno();
             uint64_t snapEnd = s->getEndSeqno();
-            if (snapStart < last_seqno && snapEnd <= last_seqno) {
+            if (snapStart < last_seqno.load() && snapEnd <= last_seqno.load()) {
                 LOG(EXTENSION_LOG_WARNING, "%s (vb %d) Erroneous snapshot "
                     "marker received, with opaque: %" PRIu32 ", its start "
                     "(%" PRIu64 "), and end (%" PRIu64 ") are less than last "
                     "received seqno (%" PRIu64 "); Dropping marker!",
                     consumer->logHeader(), vb_, opaque_, snapStart, snapEnd,
-                    last_seqno);
+                    last_seqno.load());
                 delete s;
                 return ENGINE_ERANGE;
             }
@@ -1281,12 +1281,12 @@ ENGINE_ERROR_CODE PassiveStream::processMutation(MutationResponse* mutation) {
         return ENGINE_NOT_MY_VBUCKET;
     }
 
-    if (mutation->getBySeqno() > cur_snapshot_end) {
+    if (mutation->getBySeqno() > cur_snapshot_end.load()) {
         LOG(EXTENSION_LOG_WARNING, "%s (vb %d) Erroneous mutation [sequence "
             "number (%" PRIu64 ") greater than current snapshot end seqno "
             "(%" PRIu64 ")] being processed; Dropping the mutation!",
             consumer->logHeader(), vb_, mutation->getBySeqno(),
-            cur_snapshot_end);
+            cur_snapshot_end.load());
         delete mutation;
         return ENGINE_ERANGE;
     }
@@ -1328,12 +1328,12 @@ ENGINE_ERROR_CODE PassiveStream::processDeletion(MutationResponse* deletion) {
         return ENGINE_NOT_MY_VBUCKET;
     }
 
-    if (deletion->getBySeqno() > cur_snapshot_end) {
+    if (deletion->getBySeqno() > cur_snapshot_end.load()) {
         LOG(EXTENSION_LOG_WARNING, "%s (vb %d) Erroneous deletion [sequence "
             "number (%" PRIu64 ") greater than current snapshot end seqno "
             "(%" PRIu64 ")] being processed; Dropping the deletion!",
             consumer->logHeader(), vb_, deletion->getBySeqno(),
-            cur_snapshot_end);
+            cur_snapshot_end.load());
         delete deletion;
         return ENGINE_ERANGE;
     }
@@ -1372,22 +1372,22 @@ ENGINE_ERROR_CODE PassiveStream::processDeletion(MutationResponse* deletion) {
 void PassiveStream::processMarker(SnapshotMarker* marker) {
     RCPtr<VBucket> vb = engine->getVBucket(vb_);
 
-    cur_snapshot_start = marker->getStartSeqno();
-    cur_snapshot_end = marker->getEndSeqno();
-    cur_snapshot_type = (marker->getFlags() & MARKER_FLAG_DISK) ? disk : memory;
+    cur_snapshot_start.store(marker->getStartSeqno());
+    cur_snapshot_end.store(marker->getEndSeqno());
+    cur_snapshot_type.store((marker->getFlags() & MARKER_FLAG_DISK) ? disk : memory);
 
     if (vb) {
         if (marker->getFlags() & MARKER_FLAG_DISK && vb->getHighSeqno() == 0) {
             vb->setBackfillPhase(true);
-            vb->checkpointManager.setBackfillPhase(cur_snapshot_start,
-                                                   cur_snapshot_end);
+            vb->checkpointManager.setBackfillPhase(cur_snapshot_start.load(),
+                                                   cur_snapshot_end.load());
         } else {
             if (marker->getFlags() & MARKER_FLAG_CHK ||
                 vb->checkpointManager.getOpenCheckpointId() == 0) {
-                vb->checkpointManager.createSnapshot(cur_snapshot_start,
-                                                     cur_snapshot_end);
+                vb->checkpointManager.createSnapshot(cur_snapshot_start.load(),
+                                                     cur_snapshot_end.load());
             } else {
-                vb->checkpointManager.updateCurrentSnapshotEnd(cur_snapshot_end);
+                vb->checkpointManager.updateCurrentSnapshotEnd(cur_snapshot_end.load());
             }
             vb->setBackfillPhase(false);
         }
@@ -1413,8 +1413,8 @@ void PassiveStream::processSetVBucketState(SetVBucketState* state) {
 }
 
 void PassiveStream::handleSnapshotEnd(RCPtr<VBucket>& vb, uint64_t byseqno) {
-    if (byseqno == cur_snapshot_end) {
-        if (cur_snapshot_type == disk && vb->isBackfillPhase()) {
+    if (byseqno == cur_snapshot_end.load()) {
+        if (cur_snapshot_type.load() == disk && vb->isBackfillPhase()) {
             vb->setBackfillPhase(false);
             uint64_t id = vb->checkpointManager.getOpenCheckpointId() + 1;
             vb->checkpointManager.checkAndAddNewCheckpoint(id, vb);
@@ -1438,7 +1438,7 @@ void PassiveStream::handleSnapshotEnd(RCPtr<VBucket>& vb, uint64_t byseqno) {
             }
             cur_snapshot_ack = false;
         }
-        cur_snapshot_type = none;
+        cur_snapshot_type.store(none);
     }
 }
 
@@ -1454,18 +1454,18 @@ void PassiveStream::addStats(ADD_STAT add_stat, const void *c) {
     snprintf(buf, bsize, "%s:stream_%d_items_ready", name_.c_str(), vb_);
     add_casted_stat(buf, itemsReady ? "true" : "false", add_stat, c);
     snprintf(buf, bsize, "%s:stream_%d_last_received_seqno", name_.c_str(), vb_);
-    add_casted_stat(buf, last_seqno, add_stat, c);
+    add_casted_stat(buf, last_seqno.load(), add_stat, c);
     snprintf(buf, bsize, "%s:stream_%d_ready_queue_memory", name_.c_str(), vb_);
     add_casted_stat(buf, getReadyQueueMemory(), add_stat, c);
 
     snprintf(buf, bsize, "%s:stream_%d_cur_snapshot_type", name_.c_str(), vb_);
-    add_casted_stat(buf, snapshotTypeToString(cur_snapshot_type), add_stat, c);
+    add_casted_stat(buf, snapshotTypeToString(cur_snapshot_type.load()), add_stat, c);
 
-    if (cur_snapshot_type != none) {
+    if (cur_snapshot_type.load() != none) {
         snprintf(buf, bsize, "%s:stream_%d_cur_snapshot_start", name_.c_str(), vb_);
-        add_casted_stat(buf, cur_snapshot_start, add_stat, c);
+        add_casted_stat(buf, cur_snapshot_start.load(), add_stat, c);
         snprintf(buf, bsize, "%s:stream_%d_cur_snapshot_end", name_.c_str(), vb_);
-        add_casted_stat(buf, cur_snapshot_end, add_stat, c);
+        add_casted_stat(buf, cur_snapshot_end.load(), add_stat, c);
     }
 }
 

src/dcp/stream.h

@@ -373,11 +373,12 @@ class PassiveStream : public Stream {
 
     EventuallyPersistentEngine* engine;
     DcpConsumer* consumer;
-    uint64_t last_seqno;
 
-    uint64_t cur_snapshot_start;
-    uint64_t cur_snapshot_end;
-    snapshot_type_t cur_snapshot_type;
+    AtomicValue<uint64_t> last_seqno;
+
+    AtomicValue<uint64_t> cur_snapshot_start;
+    AtomicValue<uint64_t> cur_snapshot_end;
+    AtomicValue<snapshot_type_t> cur_snapshot_type;
     bool cur_snapshot_ack;
 
     struct Buffer {