MB-35441 [SR]: Fix lock-order-inversion in PassiveStream::acceptStream

daverigby · daverigby · commit 26ee724e5a7b · 2019-08-05T15:16:50.000Z
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>
diff --git a/engines/ep/src/dcp/passive_stream.cc b/engines/ep/src/dcp/passive_stream.cc
@@ -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>(
@@ -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 {
