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_