From 8a78212bb841a7efcff74b8608b5194dc7047417 Mon Sep 17 00:00:00 2001 From: Juliet Levesque Date: Tue, 16 May 2023 14:52:11 -0700 Subject: [PATCH] [Nearby Connections] Check for shutdown before accessing ClientProxy. During the destruction of NearbyConnections, Core (which owns ClientProxy) is destructed before ServiceController (which owns EndpointManager), which means any pending tasks on the EndpointManager executor that use ClientProxy will be using garbage memory. To fix this issue, EndpointManager::DiscardEndpoint will check for an `is_shutdown` boolean set during ~EndpointManager before accessing ClientProxy. The assumption is that any accessing of ClientProxy after the destruction will be invalid. PiperOrigin-RevId: 532581122 --- connections/implementation/BUILD | 3 +- .../implementation/endpoint_manager.cc | 58 ++++++++++++++--- connections/implementation/endpoint_manager.h | 18 +++++- .../implementation/endpoint_manager_test.cc | 63 +++++++++++++++++-- internal/platform/single_thread_executor.h | 2 +- internal/platform/submittable_executor.h | 3 +- internal/test/BUILD | 2 + internal/test/fake_single_thread_executor.cc | 48 ++++++++++++++ internal/test/fake_single_thread_executor.h | 52 +++++++++++++++ 9 files changed, 231 insertions(+), 18 deletions(-) create mode 100644 internal/test/fake_single_thread_executor.cc create mode 100644 internal/test/fake_single_thread_executor.h diff --git a/connections/implementation/BUILD b/connections/implementation/BUILD index 67bd965b..0e93ee0a 100644 --- a/connections/implementation/BUILD +++ b/connections/implementation/BUILD @@ -107,13 +107,11 @@ cc_library( "//location/nearby/cpp/sharing/implementation:__pkg__", ], deps = [ - ":message_lite", "//connections:core_types", "//connections/implementation/analytics", "//connections/implementation/flags:connections_flags", "//connections/implementation/mediums", "//connections/implementation/mediums:utils", - "//connections/implementation/mediums/webrtc", "//connections/implementation/proto:offline_wire_formats_cc_proto", "//connections/v3:v3_types", "//internal/analytics:event_logger", @@ -232,6 +230,7 @@ cc_test( "//internal/platform:test_util", "//internal/platform:types", "//internal/platform/implementation/g3", # build_cleaner: keep + "//internal/test", "//proto:connections_enums_cc_proto", "@com_github_protobuf_matchers//protobuf-matchers", "@com_google_absl//absl/container:flat_hash_set", diff --git a/connections/implementation/endpoint_manager.cc b/connections/implementation/endpoint_manager.cc index 2233eab2..cd7ef40e 100644 --- a/connections/implementation/endpoint_manager.cc +++ b/connections/implementation/endpoint_manager.cc @@ -287,10 +287,16 @@ bool operator<(const EndpointManager::FrameProcessor& lhs, } EndpointManager::EndpointManager(EndpointChannelManager* manager) - : channel_manager_(manager) {} + : EndpointManager(manager, std::make_unique()) {} + +EndpointManager::EndpointManager( + EndpointChannelManager* manager, + std::unique_ptr serial_executor) + : channel_manager_(manager), serial_executor_(std::move(serial_executor)) {} EndpointManager::~EndpointManager() { NEARBY_LOG(INFO, "Initiating shutdown of EndpointManager."); + is_shutdown_ = true; analytics::ThroughputRecorderContainer::GetInstance().Shutdown(); CountDownLatch latch(1); RunOnEndpointManagerThread("bring-down-endpoints", [this, &latch]() { @@ -301,7 +307,7 @@ EndpointManager::~EndpointManager() { latch.Await(); NEARBY_LOG(INFO, "Bringing down control thread"); - serial_executor_.Shutdown(); + serial_executor_->Shutdown(); NEARBY_LOG(INFO, "EndpointManager is down"); } @@ -524,10 +530,48 @@ std::vector EndpointManager::SendPayloadChunk( void EndpointManager::DiscardEndpoint(ClientProxy* client, const std::string& endpoint_id) { NEARBY_LOGS(VERBOSE) << "DiscardEndpoint for endpoint " << endpoint_id; - RunOnEndpointManagerThread("discard-endpoint", [this, client, endpoint_id]() { - RemoveEndpoint(client, endpoint_id, - /*notify=*/client->IsConnectedToEndpoint(endpoint_id)); - }); + RunOnEndpointManagerThread( + "discard-endpoint", [this, client, endpoint_id]() { + // `ClientProxy` is destroyed before `EndpointManager` in + // `~NearbyConnections`, which means "discard-endpoint" needs to check + // if this task is being executing during `~EndpointManager` to + // prevent accessing an invalid `ClientProxy` pointer. There are two + // cases where "discard-endpoint" can be executed during destruction, + // both of which can safely use `is_shutdown_` to check if this is being + // executed during the destruction of the object: + // + // Case 1: "discard-endpoints" is posted to the thread before + // destruction, but not executed yet: `~EndpointManager` blocks on + // "bring-down-endpoints" and because the executor is a single thread + // executor, tasks are guaranteed to execute sequentially + // (see + // https://docs.oracle.com/javase/8/docs/api/java/util/concurrent/Executors.html#newSingleThreadExecutor--) + // and this means that the "discard-endpoints" will be executed before + // "bring-down-endpoints", blocking the destruction of `is_shutdown_` + // and therefore `is_shutdown_` is not garbage memory. + // + // Case 2: "discard-endpoints" is posted to the thread during + // destruction, after "bring-down-endpoints" is called: the executor + // will be destructed before `is_shutdown_` because of the ordering of + // `EndpointManager`'s member variables, and the executor's destructor + // blocks on running all pending tasks + // (see + // https://source.chromium.org/chromium/chromium/src/+/refs/heads/main:chrome/services/sharing/nearby/platform/scheduled_executor.cc;l=67;drc=e0e0d24aaa54727dc0a8bc4b159ccdf80d3f5d8d), + // which means that "discard-endpoints" will run during the destruction + // of `serial_executor_` and will still have access to a valid + // `is_shutdown_`. + // + // TODO(b/280653613): Develop a more robost solution to prevent + // accessing an already destroyed `ClientProxy` during destruction. + if (is_shutdown_) { + NEARBY_LOGS(VERBOSE) + << "DiscardEndpoint called during destruction, returning early."; + return; + } + + RemoveEndpoint(client, endpoint_id, + /*notify=*/client->IsConnectedToEndpoint(endpoint_id)); + }); } std::vector EndpointManager::SendControlMessage( @@ -698,7 +742,7 @@ void EndpointManager::EndpointState::StartEndpointKeepAliveManager( void EndpointManager::RunOnEndpointManagerThread(const std::string& name, Runnable runnable) { - serial_executor_.Execute(name, std::move(runnable)); + serial_executor_->Execute(name, std::move(runnable)); } } // namespace connections diff --git a/connections/implementation/endpoint_manager.h b/connections/implementation/endpoint_manager.h index 53c16bf0..7e7fdaae 100644 --- a/connections/implementation/endpoint_manager.h +++ b/connections/implementation/endpoint_manager.h @@ -153,6 +153,11 @@ class EndpointManager { // blocked here. void DiscardEndpoint(ClientProxy* client, const std::string& endpoint_id); + protected: + // For unit tests only to control executing tasks on the executor. + EndpointManager(EndpointChannelManager* manager, + std::unique_ptr serial_executor); + private: class EndpointState { public: @@ -288,7 +293,18 @@ class EndpointManager { // We keep track of all registered channel endpoints here. absl::flat_hash_map endpoints_; - SingleThreadExecutor serial_executor_; + // Indicates whether the destructor has been called yet. If `is_shutdown_` + // is true, assume any `ClientProxy` pointers are invalid, and should not + // be used. + // + // The ordering of these objects is important: `serial_executor_` must be + // destroyed before `is_shutdown_` because `serial_executor_` runs all + // pending tasks during it's destruction, and the "discard-endpoints" + // task checks `is_shutdown_` to prevent accessing an invalid `ClientProxy` + // pointer. + bool is_shutdown_ = false; + + std::unique_ptr serial_executor_; }; // Operator overloads when comparing FrameProcessor*. diff --git a/connections/implementation/endpoint_manager_test.cc b/connections/implementation/endpoint_manager_test.cc index df790c70..58c83162 100644 --- a/connections/implementation/endpoint_manager_test.cc +++ b/connections/implementation/endpoint_manager_test.cc @@ -34,6 +34,7 @@ #include "internal/platform/count_down_latch.h" #include "internal/platform/exception.h" #include "internal/platform/logging.h" +#include "internal/test/fake_single_thread_executor.h" #include "proto/connections_enums.pb.h" namespace nearby { @@ -111,6 +112,13 @@ class MockFrameProcessor : public EndpointManager::FrameProcessor { (override)); }; +class TestEndpointManager : public EndpointManager { + public: + TestEndpointManager(EndpointChannelManager* manager, + std::unique_ptr serial_executor) + : EndpointManager(manager, std::move(serial_executor)) {} +}; + class EndpointManagerTest : public ::testing::Test { protected: void RegisterEndpoint(std::unique_ptr channel, @@ -127,14 +135,15 @@ class EndpointManagerTest : public ::testing::Test { EXPECT_CALL(*channel, GetLastWriteTimestamp()) .WillRepeatedly(Return(start_time_)); EXPECT_CALL(mock_listener_.initiated_cb, Call).Times(1); - em_.RegisterEndpoint(&client_, endpoint_id_, info_, connection_options_, - std::move(channel), listener_, connection_token); + em_.RegisterEndpoint(client_.get(), endpoint_id_, info_, + connection_options_, std::move(channel), listener_, + connection_token); if (should_close) { EXPECT_TRUE(done.Await(absl::Milliseconds(1000)).result()); } } - ClientProxy client_; + std::unique_ptr client_ = std::make_unique(); ConnectionOptions connection_options_{ .keep_alive_interval_millis = 5000, .keep_alive_timeout_millis = 30000, @@ -195,7 +204,7 @@ TEST_F(EndpointManagerTest, UnregisterEndpointCallsOnDisconnected) { // (IMO, it should be called as long as any connection callback was called // before. (in this case initiated_cb is called)). // Test captures current protocol behavior. - em_.UnregisterEndpoint(&client_, endpoint_id_); + em_.UnregisterEndpoint(client_.get(), endpoint_id_); } TEST_F(EndpointManagerTest, RegisterFrameProcessorWorks) { @@ -250,7 +259,7 @@ TEST_F(EndpointManagerTest, UnregisterFrameProcessorWorks) { processors_.emplace_back(std::move(connect_request)); // Endpoint will not send OnDisconnect notification to frame processor. RegisterEndpoint(std::move(endpoint_channel), false); - em_.UnregisterEndpoint(&client_, endpoint_id_); + em_.UnregisterEndpoint(client_.get(), endpoint_id_); } TEST_F(EndpointManagerTest, SendControlMessageWorks) { @@ -286,7 +295,7 @@ TEST_F(EndpointManagerTest, SendControlMessageWorks) { em_.SendControlMessage(header, control, std::vector{endpoint_id_}); EXPECT_EQ(failed_ids, std::vector{}); NEARBY_LOG(INFO, "Will unregister endpoint now"); - em_.UnregisterEndpoint(&client_, endpoint_id_); + em_.UnregisterEndpoint(client_.get(), endpoint_id_); NEARBY_LOG(INFO, "Will call destructors now"); } @@ -301,6 +310,48 @@ TEST_F(EndpointManagerTest, SingleReadOnInvalidPayload) { RegisterEndpoint(std::move(endpoint_channel)); } +// Regression test for b/278729669. +// +// During the destruction of NearbyConnections, Core (which owns ClientProxy) +// is destructed before ServiceController (which owns EndpointManager), which +// means any pending tasks on the EndpointManager than use ClientProxy will +// be using garbage memory, and cause crashes. This test enforces the fix. +TEST_F(EndpointManagerTest, DisconnectEndpointDuringDestruction) { + // This test uses a `FakeSingleThreadExecutor` in order to control when + // tasks are executed in order to simulate the scenario where + // `DiscardEndpoint` is posted to the executor before the EndpointManager + // is destructed, and executed during it's destruction. + std::unique_ptr serial_executor = + std::make_unique(); + FakeSingleThreadExecutor* fake_serial_executor = + static_cast(serial_executor.get()); + std::unique_ptr endpoint_manager = + std::make_unique(&ecm_, std::move(serial_executor)); + + // DiscardEndpoint posts a task to the executor to run "discard-endpoint", + // however the `FakeSingleThreadExecutor` will not run this task + // immediately. + fake_serial_executor->SetRunExecutablesImmediately( + /*run_executables_immediately=*/false); + endpoint_manager->DiscardEndpoint(client_.get(), endpoint_id_); + + // Simulate Core destruction of ClientProxy by destroying `client_`. + client_.reset(); + + // Simulate ServiceController destruction of EndpointManager by destroying + // `endpoint_manager`, and set the `FakeSingleThreadExecutor` to run + // executables on calls `Execute`. When `endpoint_manager` is destructed, it + // will block on calls to `Execute` to run all pending executables, notably + // "discard-endpoint" from above. However, "discard-endpoint" will have a + // reference to a destroyed ClientProxy. + // + // Expect no crash when "discard-endpoints" is executed during the + // destruction. + fake_serial_executor->SetRunExecutablesImmediately( + /*run_executables_immediately=*/true); + endpoint_manager.reset(); +} + } // namespace } // namespace connections } // namespace nearby diff --git a/internal/platform/single_thread_executor.h b/internal/platform/single_thread_executor.h index 27a80f40..eb53bbee 100644 --- a/internal/platform/single_thread_executor.h +++ b/internal/platform/single_thread_executor.h @@ -24,7 +24,7 @@ namespace nearby { // queue. // // https://docs.oracle.com/javase/8/docs/api/java/util/concurrent/Executors.html#newSingleThreadExecutor-- -class ABSL_LOCKABLE SingleThreadExecutor final : public SubmittableExecutor { +class ABSL_LOCKABLE SingleThreadExecutor : public SubmittableExecutor { public: using Platform = api::ImplementationPlatform; SingleThreadExecutor() diff --git a/internal/platform/submittable_executor.h b/internal/platform/submittable_executor.h index 47aa9e80..aba0375d 100644 --- a/internal/platform/submittable_executor.h +++ b/internal/platform/submittable_executor.h @@ -17,6 +17,7 @@ #include #include +#include #include #include "absl/base/thread_annotations.h" @@ -53,7 +54,7 @@ class ABSL_LOCKABLE SubmittableExecutor : public api::SubmittableExecutor, } return *this; } - void Execute(const std::string& name, Runnable&& runnable) + virtual void Execute(const std::string& name, Runnable&& runnable) ABSL_LOCKS_EXCLUDED(mutex_) { MutexLock lock(&mutex_); if (impl_) diff --git a/internal/test/BUILD b/internal/test/BUILD index dc5407b7..d843e66c 100644 --- a/internal/test/BUILD +++ b/internal/test/BUILD @@ -18,12 +18,14 @@ cc_library( name = "test", srcs = [ "fake_clock.cc", + "fake_single_thread_executor.cc", "fake_task_runner.cc", "fake_timer.cc", ], hdrs = [ "fake_clock.h", "fake_device_info.h", + "fake_single_thread_executor.h", "fake_task_runner.h", "fake_timer.h", ], diff --git a/internal/test/fake_single_thread_executor.cc b/internal/test/fake_single_thread_executor.cc new file mode 100644 index 00000000..dc50918a --- /dev/null +++ b/internal/test/fake_single_thread_executor.cc @@ -0,0 +1,48 @@ +// Copyright 2023 Google LLC +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// https://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +#include "internal/test/fake_single_thread_executor.h" + +#include +#include +#include + +namespace nearby { + +FakeSingleThreadExecutor::FakeSingleThreadExecutor() = default; + +FakeSingleThreadExecutor::~FakeSingleThreadExecutor() { DoShutdown(); } + +void FakeSingleThreadExecutor::Execute(const std::string& name, + Runnable&& runnable) { + runnables_.push_back(std::make_pair(name, std::move(runnable))); + + if (!run_executables_immediately_) return; + + RunAllExecutables(); +} + +void FakeSingleThreadExecutor::RunAllExecutables() { + // Because `SingleThreadExecutor` ensures sequencing, run all pending + // executables in order they were added to the vector. + for (auto& runnable_pair : runnables_) { + runnable_pair.second(); + } + + runnables_.clear(); +} + +void FakeSingleThreadExecutor::DoShutdown() { RunAllExecutables(); } + +} // namespace nearby diff --git a/internal/test/fake_single_thread_executor.h b/internal/test/fake_single_thread_executor.h new file mode 100644 index 00000000..d9c86d35 --- /dev/null +++ b/internal/test/fake_single_thread_executor.h @@ -0,0 +1,52 @@ +// Copyright 2023 Google LLC +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// https://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +#ifndef PLATFORM_PUBLIC_FAKE_SINGLE_THREAD_EXECUTOR_H_ +#define PLATFORM_PUBLIC_FAKE_SINGLE_THREAD_EXECUTOR_H_ + +#include +#include +#include + +#include "absl/base/thread_annotations.h" +#include "internal/platform/single_thread_executor.h" + +namespace nearby { + +class ABSL_LOCKABLE FakeSingleThreadExecutor final + : public SingleThreadExecutor { + public: + FakeSingleThreadExecutor(); + ~FakeSingleThreadExecutor() override; + FakeSingleThreadExecutor(FakeSingleThreadExecutor&&) = default; + FakeSingleThreadExecutor& operator=(FakeSingleThreadExecutor&&) = default; + + void Execute(const std::string& name, Runnable&& runnable) override; + + void SetRunExecutablesImmediately(bool run_executables_immediately) { + run_executables_immediately_ = run_executables_immediately; + } + + void RunAllExecutables(); + + private: + void DoShutdown(); + + bool run_executables_immediately_ = false; + std::vector> runnables_; +}; + +} // namespace nearby + +#endif // PLATFORM_PUBLIC_FAKE_SINGLE_THREAD_EXECUTOR_H_