Handle possible deadlocks and hangs during PeerConnection teardown Address two issues related to thread shutdown and teardown when the target thread (network/worker) has been instructed to quit before the PeerConnection destruction has completed: 1. In ScopedOperationsBatcher::Run(), if the target thread drops tasks (e.g., because it is quitting), the loop would previously run infinitely because task_idx did not advance. We now detect this by verifying that task_idx has progressed after the BlockingCall, and abort if it has not. 2. In MethodCall::Marshal and ConstMethodCall::Marshal, if the target thread drops the task because it is quitting, the calling thread would previously hang forever because event_.Set() was never called. We now use absl::Cleanup to ensure the event is set on task destruction. Bug: b/515407473 Change-Id: Iea8ec32b6a9e9850983ed563db07b5c386ad834d Reviewed-on: https://webrtc-review.googlesource.com/c/src/+/479140 Reviewed-by: Danil Chapovalov <danilchap@webrtc.org> Commit-Queue: Tomas Gunnarsson <tommi@webrtc.org> Cr-Commit-Position: refs/heads/main@{#47930}
diff --git a/pc/BUILD.gn b/pc/BUILD.gn index 46ddb0a..b42faa8b 100644 --- a/pc/BUILD.gn +++ b/pc/BUILD.gn
@@ -52,6 +52,7 @@ "../rtc_base:rtc_event", "../rtc_base:stringutils", "../rtc_base:threading", + "//third_party/abseil-cpp/absl/cleanup", ] } @@ -3556,7 +3557,10 @@ sources = [ "scoped_operations_batcher_unittest.cc" ] deps = [ ":scoped_operations_batcher", + "../api:function_view", + "../api:location", "../api:rtc_error", + "../rtc_base:null_socket_server", "../rtc_base:threading", "../test:test_support", "//third_party/abseil-cpp/absl/functional:any_invocable",
diff --git a/pc/proxy.h b/pc/proxy.h index 7adcdd9..dfcf77c 100644 --- a/pc/proxy.h +++ b/pc/proxy.h
@@ -61,6 +61,7 @@ #include <tuple> #include <utility> +#include "absl/cleanup/cleanup.h" #include "rtc_base/event.h" #include "rtc_base/string_utils.h" // IWYU pragma: keep #include "rtc_base/thread.h" @@ -110,9 +111,11 @@ if (t->IsCurrent()) { Invoke(std::index_sequence_for<Args...>()); } else { - t->PostTask([this] { + // Use absl::Cleanup to ensure the event is set even if the task is + // dropped (e.g., because the target thread is quitting). + absl::Cleanup cleanup = [this] { event_.Set(); }; + t->PostTask([this, cleanup = std::move(cleanup)] { Invoke(std::index_sequence_for<Args...>()); - event_.Set(); }); event_.Wait(Event::kForever); } @@ -145,9 +148,11 @@ if (t->IsCurrent()) { Invoke(std::index_sequence_for<Args...>()); } else { - t->PostTask([this] { + // Use absl::Cleanup to ensure the event is set even if the task is + // dropped (e.g., because the target thread is quitting). + absl::Cleanup cleanup = [this] { event_.Set(); }; + t->PostTask([this, cleanup = std::move(cleanup)] { Invoke(std::index_sequence_for<Args...>()); - event_.Set(); }); event_.Wait(Event::kForever); }
diff --git a/pc/proxy_unittest.cc b/pc/proxy_unittest.cc index d3d3739..589b50f 100644 --- a/pc/proxy_unittest.cc +++ b/pc/proxy_unittest.cc
@@ -257,4 +257,11 @@ EXPECT_EQ("Method2", fake_proxy_->Method2(arg1, arg2)); } +TEST_F(ProxyTest, DoesNotHangWhenTargetThreadQuits) { + signaling_thread_->Stop(); + EXPECT_CALL(*fake_, Method0).Times(0); + EXPECT_CALL(*fake_, Destroy).Times(1); + fake_proxy_->Method0(); +} + } // namespace webrtc
diff --git a/pc/scoped_operations_batcher.cc b/pc/scoped_operations_batcher.cc index 61fa47f..15b25fc 100644 --- a/pc/scoped_operations_batcher.cc +++ b/pc/scoped_operations_batcher.cc
@@ -19,7 +19,6 @@ #include "api/rtc_error.h" #include "api/sequence_checker.h" #include "rtc_base/checks.h" -#include "rtc_base/logging.h" #include "rtc_base/thread.h" namespace webrtc { @@ -30,10 +29,7 @@ } ScopedOperationsBatcher::~ScopedOperationsBatcher() { - RTCError error = Run(); - if (!error.ok()) { - RTC_LOG(LS_ERROR) << "Batcher failed: " << error.message(); - } + Run(); } RTCError ScopedOperationsBatcher::Run() { @@ -58,6 +54,7 @@ RTCError error = RTCError::OK(); while (task_idx < tasks_.size()) { + size_t previous_task_idx = task_idx; target_thread_->BlockingCall([&] { while (task_idx < tasks_.size()) { if (auto* void_task = std::get_if<SimpleBatchTask>(&tasks_[task_idx])) { @@ -87,7 +84,16 @@ } } }); + // If the target thread dropped the task (e.g. because it is quitting), + // task_idx will not have advanced. Abort to prevent an infinite busy-loop. + // This is not expected to happen under normal circumstances but if the + // integrating application prematurely terminates the target thread, that's + // when it can. However, doing so, breaks design assumptions. + if (error.ok() && task_idx == previous_task_idx) { + error = RTCError::InternalError("Thread was prematurely terminated"); + } if (!error.ok()) { + RTC_LOG_ERROR(error); break; } }
diff --git a/pc/scoped_operations_batcher_unittest.cc b/pc/scoped_operations_batcher_unittest.cc index ad9d033..5d8d5ea 100644 --- a/pc/scoped_operations_batcher_unittest.cc +++ b/pc/scoped_operations_batcher_unittest.cc
@@ -15,13 +15,26 @@ #include <vector> #include "absl/functional/any_invocable.h" +#include "api/function_view.h" +#include "api/location.h" #include "api/rtc_error.h" +#include "rtc_base/null_socket_server.h" #include "rtc_base/thread.h" #include "test/gtest.h" namespace webrtc { namespace { +class FakeQuittingThread : public Thread { + public: + FakeQuittingThread() : Thread(std::make_unique<NullSocketServer>()) {} + + void BlockingCallImpl(FunctionView<void()> functor, + const Location& location) override { + // Do nothing to simulate dropping the task when quitting. + } +}; + TEST(ScopedOperationsBatcherTest, ExecutesTasksOnTargetThread) { auto target_thread = Thread::Create(); target_thread->Start(); @@ -164,5 +177,26 @@ EXPECT_TRUE(batcher.IsEmpty()); } +TEST(ScopedOperationsBatcherTest, AbortsGracefullyWhenTargetThreadQuits) { + FakeQuittingThread target_thread; + target_thread.Start(); + + ScopedOperationsBatcher batcher(&target_thread); + batcher.Add([] {}); + + // This should return an error and not hang or crash. + RTCError error = batcher.Run(); + EXPECT_FALSE(error.ok()); + EXPECT_EQ(error.type(), RTCErrorType::INTERNAL_ERROR); + + // Explicitly stop `target_thread` before it is destroyed to avoid a conflict + // between the FakeQuittingThread destructor and ~Thread(). + // Otherwise, the background thread continues running `ProcessMessages()` and + // making virtual calls (like `IsQuitting()`) while the + // `~FakeQuittingThread()` destructor is run and resets the vptr (which + // triggers a race with TSAN). + target_thread.Stop(); +} + } // namespace } // namespace webrtc