Skip to content

Commit

Permalink
MB-35441 [SR]: Fix lock-order-inversion in PassiveStream::acceptStream
Browse files Browse the repository at this point in the history 
As identified by TSan, triggered by http://review.couchbase.org/#/c/112781/ (fix for MB-35361):

    $ TSAN_OPTIONS=second_deadlock_stack=1 ./ep-engine_ep_unit_tests --gtest_filter=AllBucketTypes/DurabilityPassiveStreamTest.ReceiveMutationInsteadOfCommitForReconnectWindowWithPrepareLast/ephemeral_auto_delete
    Note: Google Test filter = AllBucketTypes/DurabilityPassiveStreamTest.ReceiveMutationInsteadOfCommitForReconnectWindowWithPrepareLast/ephemeral_auto_delete
    [==========] Running 1 test from 1 test case.
    [----------] Global test environment set-up.
    [----------] 1 test from AllBucketTypes/DurabilityPassiveStreamTest
    [ RUN      ] AllBucketTypes/DurabilityPassiveStreamTest.ReceiveMutationInsteadOfCommitForReconnectWindowWithPrepareLast/ephemeral_auto_delete
    ==================
    WARNING: ThreadSanitizer: lock-order-inversion (potential deadlock) (pid=55122)
    Cycle in lock order graph: M458376862792418680 (0x000000000000) => M431917940103981768 (0x000000000000) => M456969213031230168 (0x000000000000) => M458376862792418680

      Mutex M431917940103981768 acquired here while holding mutex M458376862792418680 in main thread:
	#0 AnnotateRWLockAcquired <null> (libtsan.so.0+0x00000005ea8d)
	...
	#10 folly::SynchronizedPtr<...>::rlock() const folly/SynchronizedPtr.h:86 (ep-engine_ep_unit_tests+0x000000ea13b0)
	#11 PassiveDurabilityMonitor::getHighPreparedSeqno() const kv_engine/engines/ep/src/durability/passive_durability_monitor.cc:115 (ep-engine_ep_unit_tests+0x000000ea13b0)
	#12 VBucket::getHighPreparedSeqno() const kv_engine/engines/ep/src/vbucket.cc:285 (ep-engine_ep_unit_tests+0x000000f9e1f4)
	#13 PassiveStream::acceptStream(cb::mcbp::Status, unsigned int) kv_engine/engines/ep/src/dcp/passive_stream.cc:176 (ep-engine_ep_unit_tests+0x000000e5995d)
	#14 DurabilityPassiveStreamTest::testReceiveMutationOrDeletionInsteadOfCommitForReconnectWindowWithPrepareLast(DocumentState) kv_engine/engines/ep/tests/module_tests/dcp_durability_stream_test.cc:929 (ep-engine_ep_unit_tests+0x0000006c7905)
	...

      Mutex M458376862792418680 previously acquired by the same thread here:
	#0 pthread_mutex_lock <null> (libtsan.so.0+0x00000003bbbf)
	...
	couchbase#5 PassiveStream::acceptStream(cb::mcbp::Status, unsigned int) kv_engine/engines/ep/src/dcp/passive_stream.cc:162 (ep-engine_ep_unit_tests+0x000000e59789)
	couchbase#6 DurabilityPassiveStreamTest::testReceiveMutationOrDeletionInsteadOfCommitForReconnectWindowWithPrepareLast(DocumentState) kv_engine/engines/ep/tests/module_tests/dcp_durability_stream_test.cc:929 (ep-engine_ep_unit_tests+0x0000006c7905)
	...

      Mutex M456969213031230168 acquired here while holding mutex M431917940103981768 in main thread:
	#0 AnnotateRWLockAcquired <null> (libtsan.so.0+0x00000005ea8d)
	...
	#9 folly::SynchronizedBase<...>::wlock() folly/Synchronized.h:114 (ep-engine_ep_unit_tests+0x000000ea3db0)
	#10 PassiveDurabilityMonitor::storeSeqnoAck(long, long) kv_engine/engines/ep/src/durability/passive_durability_monitor.cc:244 (ep-engine_ep_unit_tests+0x000000ea3db0)
	#11 PassiveDurabilityMonitor::notifySnapshotEndReceived(unsigned long) kv_engine/engines/ep/src/durability/passive_durability_monitor.cc:219 (ep-engine_ep_unit_tests+0x000000ea3db0)
	#12 VBucket::notifyPassiveDMOfSnapEndReceived(unsigned long) kv_engine/engines/ep/src/vbucket.cc:979 (ep-engine_ep_unit_tests+0x000000f9e8b8)
	#13 PassiveStream::handleSnapshotEnd(std::shared_ptr<VBucket>&, unsigned long) kv_engine/engines/ep/src/dcp/passive_stream.cc:978 (ep-engine_ep_unit_tests+0x000000e51c63)
	#14 PassiveStream::processMessage(MutationConsumerMessage*, PassiveStream::MessageType) kv_engine/engines/ep/src/dcp/passive_stream.cc:662 (ep-engine_ep_unit_tests+0x000000e55910)
	#15 PassiveStream::processPrepare(MutationConsumerMessage*) kv_engine/engines/ep/src/dcp/passive_stream.cc:685 (ep-engine_ep_unit_tests+0x000000e55e29)
	#16 PassiveStream::messageReceived(...) kv_engine/engines/ep/src/dcp/passive_stream.cc:311 (ep-engine_ep_unit_tests+0x000000e59f8c)
	#17 MockPassiveStream::messageReceived(...) kv_engine/engines/ep/tests/mock/mock_stream.cc:87 (ep-engine_ep_unit_tests+0x0000004587a2)
	#18 DurabilityPassiveStreamTest::makeAndReceiveDcpPrepare(...) kv_engine/engines/ep/tests/module_tests/dcp_durability_stream_test.cc:1340 (ep-engine_ep_unit_tests+0x0000006bcbda)
	#19 DurabilityPassiveStreamTest::testReceiveMutationOrDeletionInsteadOfCommitForReconnectWindowWithPrepareLast(DocumentState) kv_engine/engines/ep/tests/module_tests/dcp_durability_stream_test.cc:920 (ep-engine_ep_unit_tests+0x0000006c7748)
	...

      Mutex M431917940103981768 previously acquired by the same thread here:
	#0 AnnotateRWLockAcquired <null> (libtsan.so.0+0x00000005ea8d)
	...
	#10 folly::SynchronizedPtr<...>::wlock() folly/SynchronizedPtr.h:95 (ep-engine_ep_unit_tests+0x000000ea3b94)
	#11 PassiveDurabilityMonitor::notifySnapshotEndReceived(unsigned long) kv_engine/engines/ep/src/durability/passive_durability_monitor.cc:208 (ep-engine_ep_unit_tests+0x000000ea3b94)
	#12 VBucket::notifyPassiveDMOfSnapEndReceived(unsigned long) kv_engine/engines/ep/src/vbucket.cc:979 (ep-engine_ep_unit_tests+0x000000f9e8b8)
	#13 PassiveStream::handleSnapshotEnd(std::shared_ptr<VBucket>&, unsigned long) kv_engine/engines/ep/src/dcp/passive_stream.cc:978 (ep-engine_ep_unit_tests+0x000000e51c63)
	#14 PassiveStream::processMessage(MutationConsumerMessage*, PassiveStream::MessageType) kv_engine/engines/ep/src/dcp/passive_stream.cc:662 (ep-engine_ep_unit_tests+0x000000e55910)
	#15 PassiveStream::processPrepare(MutationConsumerMessage*) kv_engine/engines/ep/src/dcp/passive_stream.cc:685 (ep-engine_ep_unit_tests+0x000000e55e29)
	#16 PassiveStream::messageReceived(...) kv_engine/engines/ep/src/dcp/passive_stream.cc:311 (ep-engine_ep_unit_tests+0x000000e59f8c)
	#17 MockPassiveStream::messageReceived(...) kv_engine/engines/ep/tests/mock/mock_stream.cc:87 (ep-engine_ep_unit_tests+0x0000004587a2)
	#18 DurabilityPassiveStreamTest::makeAndReceiveDcpPrepare(...) kv_engine/engines/ep/tests/module_tests/dcp_durability_stream_test.cc:1340 (ep-engine_ep_unit_tests+0x0000006bcbda)
	#19 DurabilityPassiveStreamTest::testReceiveMutationOrDeletionInsteadOfCommitForReconnectWindowWithPrepareLast(DocumentState) kv_engine/engines/ep/tests/module_tests/dcp_durability_stream_test.cc:920 (ep-engine_ep_unit_tests+0x0000006c7748)
	...

      Mutex M458376862792418680 acquired here while holding mutex M456969213031230168 in main thread:
	#0 pthread_mutex_lock <null> (libtsan.so.0+0x00000003bbbf)
	...
	couchbase#4 PassiveStream::seqnoAck(long) kv_engine/engines/ep/src/dcp/passive_stream.cc:693 (ep-engine_ep_unit_tests+0x000000e51553)
	couchbase#5 DcpConsumer::seqnoAckStream(Vbid, long) kv_engine/engines/ep/src/dcp/consumer.cc:1067 (ep-engine_ep_unit_tests+0x000000e31c7c)
	couchbase#6 DcpConnMap::seqnoAckVBPassiveStream(Vbid, long) kv_engine/engines/ep/src/dcp/dcpconnmap.cc:466 (ep-engine_ep_unit_tests+0x000000e3b001)
	#7 operator() kv_engine/engines/ep/src/kv_bucket.cc:2680 (ep-engine_ep_unit_tests+0x000000f41376)
	#8 _M_invoke /usr/local/include/c++/7.3.0/bits/std_function.h:316 (ep-engine_ep_unit_tests+0x000000f41376)
	#9 std::function<void (Vbid, long)>::operator()(Vbid, long) const /usr/local/include/c++/7.3.0/bits/std_function.h:706 (ep-engine_ep_unit_tests+0x000000f9e9aa)
	#10 VBucket::sendSeqnoAck(long) kv_engine/engines/ep/src/vbucket.cc:984 (ep-engine_ep_unit_tests+0x000000f9e9aa)
	#11 PassiveDurabilityMonitor::sendSeqnoAck() kv_engine/engines/ep/src/durability/passive_durability_monitor.cc:255 (ep-engine_ep_unit_tests+0x000000ea1d60)
	#12 PassiveDurabilityMonitor::notifySnapshotEndReceived(unsigned long) kv_engine/engines/ep/src/durability/passive_durability_monitor.cc:226 (ep-engine_ep_unit_tests+0x000000ea3e27)
	#13 VBucket::notifyPassiveDMOfSnapEndReceived(unsigned long) kv_engine/engines/ep/src/vbucket.cc:979 (ep-engine_ep_unit_tests+0x000000f9e8b8)
	#14 PassiveStream::handleSnapshotEnd(std::shared_ptr<VBucket>&, unsigned long) kv_engine/engines/ep/src/dcp/passive_stream.cc:978 (ep-engine_ep_unit_tests+0x000000e51c63)
	#15 PassiveStream::processMessage(MutationConsumerMessage*, PassiveStream::MessageType) kv_engine/engines/ep/src/dcp/passive_stream.cc:662 (ep-engine_ep_unit_tests+0x000000e55910)
	#16 PassiveStream::processMutation(MutationConsumerMessage*) kv_engine/engines/ep/src/dcp/passive_stream.cc:670 (ep-engine_ep_unit_tests+0x000000e55d32)
	#17 MockPassiveStream::processMutation(MutationConsumerMessage*) <null> (ep-engine_ep_unit_tests+0x0000004596dc)
	#18 PassiveStream::messageReceived(...) kv_engine/engines/ep/src/dcp/passive_stream.cc:299 (ep-engine_ep_unit_tests+0x000000e59fdf)
	#19 MockPassiveStream::messageReceived(...) kv_engine/engines/ep/tests/mock/mock_stream.cc:87 (ep-engine_ep_unit_tests+0x0000004587a2)
	#20 DurabilityPassiveStreamTest::testReceiveMutationOrDeletionInsteadOfCommitForReconnectWindowWithPrepareLast(DocumentState) kv_engine/engines/ep/tests/module_tests/dcp_durability_stream_test.cc:950 (ep-engine_ep_unit_tests+0x0000006c7dfa)
	#21 DurabilityPassiveStreamTest_ReceiveMutationInsteadOfCommitForReconnectWindowWithPrepareLast_Test::TestBody() kv_engine/engines/ep/tests/module_tests/dcp_durability_stream_test.cc:1012 (ep-engine_ep_unit_tests+0x0000006c9970)
	...

      Mutex M456969213031230168 previously acquired by the same thread here:
	#0 AnnotateRWLockAcquired <null> (libtsan.so.0+0x00000005ea8d)
	...
	#9 folly::SynchronizedBase<...>::wlock() folly/Synchronized.h:114 (ep-engine_ep_unit_tests+0x000000ea1d32)
	#10 PassiveDurabilityMonitor::sendSeqnoAck() kv_engine/engines/ep/src/durability/passive_durability_monitor.cc:253 (ep-engine_ep_unit_tests+0x000000ea1d32)
	#11 PassiveDurabilityMonitor::notifySnapshotEndReceived(unsigned long) kv_engine/engines/ep/src/durability/passive_durability_monitor.cc:226 (ep-engine_ep_unit_tests+0x000000ea3e27)
	#12 VBucket::notifyPassiveDMOfSnapEndReceived(unsigned long) kv_engine/engines/ep/src/vbucket.cc:979 (ep-engine_ep_unit_tests+0x000000f9e8b8)
	#13 PassiveStream::handleSnapshotEnd(std::shared_ptr<VBucket>&, unsigned long) kv_engine/engines/ep/src/dcp/passive_stream.cc:978 (ep-engine_ep_unit_tests+0x000000e51c63)
	#14 PassiveStream::processMessage(MutationConsumerMessage*, PassiveStream::MessageType) kv_engine/engines/ep/src/dcp/passive_stream.cc:662 (ep-engine_ep_unit_tests+0x000000e55910)
	#15 PassiveStream::processMutation(MutationConsumerMessage*) kv_engine/engines/ep/src/dcp/passive_stream.cc:670 (ep-engine_ep_unit_tests+0x000000e55d32)
	#16 MockPassiveStream::processMutation(MutationConsumerMessage*) <null> (ep-engine_ep_unit_tests+0x0000004596dc)
	#17 PassiveStream::messageReceived(...) kv_engine/engines/ep/src/dcp/passive_stream.cc:299 (ep-engine_ep_unit_tests+0x000000e59fdf)
	#18 MockPassiveStream::messageReceived(...) kv_engine/engines/ep/tests/mock/mock_stream.cc:87 (ep-engine_ep_unit_tests+0x0000004587a2)
	#19 DurabilityPassiveStreamTest::testReceiveMutationOrDeletionInsteadOfCommitForReconnectWindowWithPrepareLast(DocumentState) kv_engine/engines/ep/tests/module_tests/dcp_durability_stream_test.cc:950 (ep-engine_ep_unit_tests+0x0000006c7dfa)
	...

Change-Id: I9a672be7c04ff2993f524ccd961ddd40ef004c17
Reviewed-on: http://review.couchbase.org/112908
Reviewed-by: James Harrison <james.harrison@couchbase.com>
Reviewed-by: Ben Huddleston <ben.huddleston@couchbase.com>
Tested-by: Build Bot <build@couchbase.com>
  • Loading branch information
@daverigby
daverigby committed Aug 5, 2019
1 parent 32119d4 commit 26ee724
Showing 1 changed file with 29 additions and 9 deletions.
38 changes: 29 additions & 9 deletions engines/ep/src/dcp/passive_stream.cc
View file
Original file line number Diff line number Diff line change
Expand Up @@ -159,6 +159,32 @@ std::string PassiveStream::getStreamTypeName() const {
}

void PassiveStream::acceptStream(cb::mcbp::Status status, uint32_t add_opaque) {
VBucketPtr vb = engine->getVBucket(vb_);
if (!vb) {
log(spdlog::level::level_enum::warn,
"({}) PassiveStream::acceptStream(): status:{} - Unable to find "
"VBucket - cannot accept Stream.",
vb_,
status);
return;
}

auto consumer = consumerPtr.lock();
if (!consumer) {
log(spdlog::level::level_enum::warn,
"({}) PassiveStream::acceptStream(): status:{} - Unable to lock "
"Consumer - cannot accept Stream.",
vb_,
status);
return;
}

// For SyncReplication streams lookup the highPreparedeSqno to check if
// we need to re-ACK (after accepting the stream).
const int64_t highPreparedSeqno = consumer->isSyncReplicationEnabled()
? vb->getHighPreparedSeqno()
: 0;

std::unique_lock<std::mutex> lh(streamMutex);
if (isPending()) {
pushToReadyQ(std::make_unique<AddStreamResponse>(
Expand All @@ -169,15 +195,9 @@ void PassiveStream::acceptStream(cb::mcbp::Status status, uint32_t add_opaque) {
// replica can commit any in-flight SyncWrites if no further
// SyncWrites are done and no disk snapshots processed by this
// replica.
{
auto consumer = consumerPtr.lock();
if (consumer && consumer->isSyncReplicationEnabled()) {
auto vb = engine->getVBucket(vb_);
if (vb && vb->getHighPreparedSeqno() != 0) {
pushToReadyQ(std::make_unique<SeqnoAcknowledgement>(
opaque_, vb_, vb->getHighPreparedSeqno()));
}
}
if (highPreparedSeqno) {
pushToReadyQ(std::make_unique<SeqnoAcknowledgement>(
opaque_, vb_, highPreparedSeqno));
}
transitionState(StreamState::AwaitingFirstSnapshotMarker);
} else {
Expand Down

0 comments on commit 26ee724

Please sign in to comment.