Skip to content

Commit 8f529bf

Browse files
abhinavdangetichiyoung
abhinavdangeti
authored and
chiyoung
committed
MB-19405: [BP] 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) #1 DcpConsumer::doRollback(unsigned int, unsigned short, unsigned long) /home/abhinav/couchbase/ep-engine/src/dcp/consumer.cc:676 (ep.so+0x00000005db67) #2 RollbackTask::run() /home/abhinav/couchbase/ep-engine/src/dcp/consumer.cc:574 (ep.so+0x00000005d9d4) #3 ExecutorThread::run() /home/abhinav/couchbase/ep-engine/src/executorthread.cc:112 (ep.so+0x0000000f8916) #4 launch_executor_thread(void*) /home/abhinav/couchbase/ep-engine/src/executorthread.cc:33 (ep.so+0x0000000f84b5) #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) #1 PassiveStream::setDead(end_stream_status_t) /home/abhinav/couchbase/ep-engine/src/dcp/stream.cc:1056 (ep.so+0x0000000766d7) #2 DcpConsumer::closeAllStreams() /home/abhinav/couchbase/ep-engine/src/dcp/consumer.cc:860 (ep.so+0x00000005a006) #3 DcpConnMap::disconnect_UNLOCKED(void const*) /home/abhinav/couchbase/ep-engine/src/connmap.cc:1137 (ep.so+0x000000049972) #4 DcpConnMap::disconnect(void const*) /home/abhinav/couchbase/ep-engine/src/connmap.cc:1111 (ep.so+0x00000004969b) #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) (Reviewed-on: http://review.couchbase.org/55785) Change-Id: I287bd95f8b03cb207419d0a0e57ca71be6058b19 Reviewed-on: http://review.couchbase.org/63446 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>
1 parent 4de869b commit 8f529bf

2 files changed

Lines changed: 30 additions & 28 deletions

File tree

‎src/dcp-stream.cc‎

Lines changed: 26 additions & 24 deletions
Original file line numberDiff line numberDiff line change
@@ -1155,7 +1155,7 @@ uint32_t PassiveStream::setDead(end_stream_status_t status) {
11551155
uint32_t unackedBytes = clearBuffer();
11561156
LOG(EXTENSION_LOG_WARNING, "%s (vb %d) Setting stream to dead state,"
11571157
" last_seqno is %llu, unackedBytes is %u, status is %s",
1158-
consumer->logHeader(), vb_, last_seqno, unackedBytes,
1158+
consumer->logHeader(), vb_, last_seqno.load(), unackedBytes,
11591159
getEndStreamStatusStr(status));
11601160
return unackedBytes;
11611161
}
@@ -1189,7 +1189,7 @@ void PassiveStream::reconnectStream(RCPtr<VBucket> &vb,
11891189
start_seqno, end_seqno_, snap_start_seqno_, snap_end_seqno_);
11901190

11911191
LockHolder lh(streamMutex);
1192-
last_seqno = start_seqno;
1192+
last_seqno.store(start_seqno);
11931193
pushToReadyQ(new StreamRequest(vb_, new_opaque, flags_, start_seqno,
11941194
end_seqno_, vb_uuid_, snap_start_seqno_,
11951195
snap_end_seqno_));
@@ -1216,29 +1216,29 @@ ENGINE_ERROR_CODE PassiveStream::messageReceived(DcpResponse* resp) {
12161216
{
12171217
MutationResponse* m = static_cast<MutationResponse*>(resp);
12181218
uint64_t bySeqno = m->getBySeqno();
1219-
if (bySeqno <= last_seqno) {
1219+
if (bySeqno <= last_seqno.load()) {
12201220
LOG(EXTENSION_LOG_WARNING, "%s (vb %d) Erroneous (out of "
12211221
"sequence) mutation received, with opaque: %ld, its "
12221222
"seqno (%llu) is not greater than last received seqno "
12231223
"(%llu); Dropping mutation!", consumer->logHeader(),
1224-
vb_, opaque_, bySeqno, last_seqno);
1224+
vb_, opaque_, bySeqno, last_seqno.load());
12251225
delete m;
12261226
return ENGINE_ERANGE;
12271227
}
1228-
last_seqno = bySeqno;
1228+
last_seqno.store(bySeqno);
12291229
break;
12301230
}
12311231
case DCP_SNAPSHOT_MARKER:
12321232
{
12331233
SnapshotMarker* s = static_cast<SnapshotMarker*>(resp);
12341234
uint64_t snapStart = s->getStartSeqno();
12351235
uint64_t snapEnd = s->getEndSeqno();
1236-
if (snapStart < last_seqno && snapEnd <= last_seqno) {
1236+
if (snapStart < last_seqno.load() && snapEnd <= last_seqno.load()) {
12371237
LOG(EXTENSION_LOG_WARNING, "%s (vb %d) Erroneous snapshot "
12381238
"marker received, with opaque: %ld, its start (%llu), and"
12391239
"end (%llu) are less than last received seqno (%llu); "
12401240
"Dropping marker!", consumer->logHeader(), vb_, opaque_,
1241-
snapStart, snapEnd, last_seqno);
1241+
snapStart, snapEnd, last_seqno.load());
12421242
delete s;
12431243
return ENGINE_ERANGE;
12441244
}
@@ -1333,19 +1333,20 @@ ENGINE_ERROR_CODE PassiveStream::processMutation(MutationResponse* mutation) {
13331333
return ENGINE_NOT_MY_VBUCKET;
13341334
}
13351335

1336-
if (mutation->getBySeqno() > cur_snapshot_end) {
1336+
if (mutation->getBySeqno() > cur_snapshot_end.load()) {
13371337
LOG(EXTENSION_LOG_WARNING, "%s (vb %d) Erroneous mutation [sequence "
13381338
"number (%llu) greater than current snapshot end seqno (%llu)] "
13391339
"being processed; Dropping the mutation!", consumer->logHeader(),
1340-
vb_, mutation->getBySeqno(), cur_snapshot_end);
1340+
vb_, mutation->getBySeqno(), cur_snapshot_end.load());
13411341
return ENGINE_ERANGE;
13421342
}
13431343

13441344
ENGINE_ERROR_CODE ret;
13451345
if (saveSnapshot) {
13461346
LockHolder lh = vb->getSnapshotLock();
13471347
ret = commitMutation(mutation, vb->isBackfillPhase());
1348-
vb->setCurrentSnapshot_UNLOCKED(cur_snapshot_start, cur_snapshot_end);
1348+
vb->setCurrentSnapshot_UNLOCKED(cur_snapshot_start.load(),
1349+
cur_snapshot_end.load());
13491350
saveSnapshot = false;
13501351
lh.unlock();
13511352
} else {
@@ -1385,19 +1386,20 @@ ENGINE_ERROR_CODE PassiveStream::processDeletion(MutationResponse* deletion) {
13851386
return ENGINE_NOT_MY_VBUCKET;
13861387
}
13871388

1388-
if (deletion->getBySeqno() > cur_snapshot_end) {
1389+
if (deletion->getBySeqno() > cur_snapshot_end.load()) {
13891390
LOG(EXTENSION_LOG_WARNING, "%s (vb %d) Erroneous deletion [sequence "
13901391
"number (%llu) greater than current snapshot end seqno (%llu)] "
13911392
"being processed; Dropping the deletion!", consumer->logHeader(),
1392-
vb_, deletion->getBySeqno(), cur_snapshot_end);
1393+
vb_, deletion->getBySeqno(), cur_snapshot_end.load());
13931394
return ENGINE_ERANGE;
13941395
}
13951396

13961397
ENGINE_ERROR_CODE ret;
13971398
if (saveSnapshot) {
13981399
LockHolder lh = vb->getSnapshotLock();
13991400
ret = commitDeletion(deletion, vb->isBackfillPhase());
1400-
vb->setCurrentSnapshot_UNLOCKED(cur_snapshot_start, cur_snapshot_end);
1401+
vb->setCurrentSnapshot_UNLOCKED(cur_snapshot_start.load(),
1402+
cur_snapshot_end.load());
14011403
saveSnapshot = false;
14021404
lh.unlock();
14031405
} else {
@@ -1435,9 +1437,9 @@ ENGINE_ERROR_CODE PassiveStream::commitDeletion(MutationResponse* deletion,
14351437
void PassiveStream::processMarker(SnapshotMarker* marker) {
14361438
RCPtr<VBucket> vb = engine->getVBucket(vb_);
14371439

1438-
cur_snapshot_start = marker->getStartSeqno();
1439-
cur_snapshot_end = marker->getEndSeqno();
1440-
cur_snapshot_type = (marker->getFlags() & MARKER_FLAG_DISK) ? disk : memory;
1440+
cur_snapshot_start.store(marker->getStartSeqno());
1441+
cur_snapshot_end.store(marker->getEndSeqno());
1442+
cur_snapshot_type.store((marker->getFlags() & MARKER_FLAG_DISK) ? disk : memory);
14411443
saveSnapshot = true;
14421444

14431445
if (vb) {
@@ -1474,8 +1476,8 @@ void PassiveStream::processSetVBucketState(SetVBucketState* state) {
14741476
}
14751477

14761478
void PassiveStream::handleSnapshotEnd(RCPtr<VBucket>& vb, uint64_t byseqno) {
1477-
if (byseqno == cur_snapshot_end) {
1478-
if (cur_snapshot_type == disk && vb->isBackfillPhase()) {
1479+
if (byseqno == cur_snapshot_end.load()) {
1480+
if (cur_snapshot_type.load() == disk && vb->isBackfillPhase()) {
14791481
vb->setBackfillPhase(false);
14801482
uint64_t id = vb->checkpointManager.getOpenCheckpointId() + 1;
14811483
vb->checkpointManager.checkAndAddNewCheckpoint(id, vb);
@@ -1499,7 +1501,7 @@ void PassiveStream::handleSnapshotEnd(RCPtr<VBucket>& vb, uint64_t byseqno) {
14991501
}
15001502
cur_snapshot_ack = false;
15011503
}
1502-
cur_snapshot_type = none;
1504+
cur_snapshot_type.store(none);
15031505
vb->setCurrentSnapshot(byseqno, byseqno);
15041506
}
15051507
}
@@ -1523,18 +1525,18 @@ void PassiveStream::addStats(ADD_STAT add_stat, const void *c) {
15231525
snprintf(buf, bsize, "%s:stream_%d_items_ready", name_.c_str(), vb_);
15241526
add_casted_stat(buf, itemsReady.load() ? "true" : "false", add_stat, c);
15251527
snprintf(buf, bsize, "%s:stream_%d_last_received_seqno", name_.c_str(), vb_);
1526-
add_casted_stat(buf, last_seqno, add_stat, c);
1528+
add_casted_stat(buf, last_seqno.load(), add_stat, c);
15271529
snprintf(buf, bsize, "%s:stream_%d_ready_queue_memory", name_.c_str(), vb_);
15281530
add_casted_stat(buf, getReadyQueueMemory(), add_stat, c);
15291531

15301532
snprintf(buf, bsize, "%s:stream_%d_cur_snapshot_type", name_.c_str(), vb_);
1531-
add_casted_stat(buf, snapshotTypeToString(cur_snapshot_type), add_stat, c);
1533+
add_casted_stat(buf, snapshotTypeToString(cur_snapshot_type.load()), add_stat, c);
15321534

1533-
if (cur_snapshot_type != none) {
1535+
if (cur_snapshot_type.load() != none) {
15341536
snprintf(buf, bsize, "%s:stream_%d_cur_snapshot_start", name_.c_str(), vb_);
1535-
add_casted_stat(buf, cur_snapshot_start, add_stat, c);
1537+
add_casted_stat(buf, cur_snapshot_start.load(), add_stat, c);
15361538
snprintf(buf, bsize, "%s:stream_%d_cur_snapshot_end", name_.c_str(), vb_);
1537-
add_casted_stat(buf, cur_snapshot_end, add_stat, c);
1539+
add_casted_stat(buf, cur_snapshot_end.load(), add_stat, c);
15381540
}
15391541
}
15401542

‎src/dcp-stream.h‎

Lines changed: 4 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -428,11 +428,11 @@ class PassiveStream : public Stream {
428428

429429
EventuallyPersistentEngine* engine;
430430
dcp_consumer_t consumer;
431-
uint64_t last_seqno;
431+
AtomicValue<uint64_t> last_seqno;
432432

433-
uint64_t cur_snapshot_start;
434-
uint64_t cur_snapshot_end;
435-
snapshot_type_t cur_snapshot_type;
433+
AtomicValue<uint64_t> cur_snapshot_start;
434+
AtomicValue<uint64_t> cur_snapshot_end;
435+
AtomicValue<snapshot_type_t> cur_snapshot_type;
436436
bool cur_snapshot_ack;
437437
bool saveSnapshot;
438438

0 commit comments

Comments
 (0)