[mojo-public] Async dispatch for associated errors

Associated endpoints have been getting error handler invocations
synchronously whenever their corresponding master interface endpoint is
closed. This is bad for reasons explained in the linked bug. Fix!

Bug: 864731
Change-Id: If4a7d205ab42be5ee43926be619e77ad0ab15d68
Reviewed-on: https://chromium-review.googlesource.com/1141078Reviewed-by: Daniel Murphy <dmurph@chromium.org>
Reviewed-by: Reilly Grant <reillyg@chromium.org>
Commit-Queue: Ken Rockot <rockot@chromium.org>
Cr-Commit-Position: refs/heads/master@{#576288}

content/browser/dom_storage/local_storage_context_mojo_unittest.cc

@@ -805,6 +805,15 @@ class LocalStorageContextMojoTestWithService
   }
 
  private:
+  // testing::Test:
+  void TearDown() override {
+    // Some of these tests close message pipes which serve as master interfaces
+    // to other associated interfaces; this in turn schedules tasks to invoke
+    // the associated interfaces' error handlers, and local storage code relies
+    // on those handlers running in order to avoid memory leaks at shutdown.
+    RunUntilIdle();
+  }
+
   DISALLOW_COPY_AND_ASSIGN(LocalStorageContextMojoTestWithService);
 };
 

content/renderer/dom_storage/local_storage_cached_areas_unittest.cc

@@ -11,12 +11,25 @@
 #include "content/public/common/content_features.h"
 #include "content/renderer/dom_storage/local_storage_cached_area.h"
 #include "content/renderer/dom_storage/mock_leveldb_wrapper.h"
+#include "testing/gtest/include/gtest/gtest.h"
 #include "third_party/blink/public/platform/scheduler/test/fake_renderer_scheduler.h"
 
 namespace content {
 
-TEST(LocalStorageCachedAreasTest, CacheLimit) {
-  base::test::ScopedTaskEnvironment scoped_task_environment;
+class LocalStorageCachedAreasTest : public testing::Test {
+  // testing::Test:
+  void TearDown() override {
+    // Some of these tests close message pipes which serve as master interfaces
+    // to other associated interfaces; this in turn schedules tasks to invoke
+    // the associated interfaces' error handlers, and local storage code relies
+    // on those handlers running in order to avoid memory leaks at shutdown.
+    scoped_task_environment_.RunUntilIdle();
+  }
+
+  base::test::ScopedTaskEnvironment scoped_task_environment_;
+};
+
+TEST_F(LocalStorageCachedAreasTest, CacheLimit) {
   const url::Origin kOrigin = url::Origin::Create(GURL("http://dom_storage1/"));
   const url::Origin kOrigin2 =
       url::Origin::Create(GURL("http://dom_storage2/"));
@@ -59,8 +72,7 @@ TEST(LocalStorageCachedAreasTest, CacheLimit) {
   EXPECT_EQ(cached_area2->memory_used(), cached_areas.TotalCacheSize());
 }
 
-TEST(LocalStorageCachedAreasTest, CloneBeforeGetArea) {
-  base::test::ScopedTaskEnvironment scoped_task_environment;
+TEST_F(LocalStorageCachedAreasTest, CloneBeforeGetArea) {
   base::test::ScopedFeatureList feature_list;
   feature_list.InitAndEnableFeature(features::kMojoSessionStorage);
   const std::string kNamespace1 = base::GenerateGUID();

mojo/public/cpp/bindings/lib/multiplex_router.cc

@@ -338,8 +338,9 @@ MultiplexRouter::MultiplexRouter(
     connector_.AllowWokenUpBySyncWatchOnSameThread();
   }
   connector_.set_incoming_receiver(&filters_);
-  connector_.set_connection_error_handler(base::Bind(
-      &MultiplexRouter::OnPipeConnectionError, base::Unretained(this)));
+  connector_.set_connection_error_handler(
+      base::BindOnce(&MultiplexRouter::OnPipeConnectionError,
+                     base::Unretained(this), false /* force_async_dispatch */));
 
   std::unique_ptr<MessageHeaderValidator> header_validator =
       std::make_unique<MessageHeaderValidator>();
@@ -491,7 +492,7 @@ void MultiplexRouter::RaiseError() {
     connector_.RaiseError();
   } else {
     task_runner_->PostTask(FROM_HERE,
-                           base::Bind(&MultiplexRouter::RaiseError, this));
+                           base::BindOnce(&MultiplexRouter::RaiseError, this));
   }
 }
 
@@ -506,7 +507,7 @@ void MultiplexRouter::CloseMessagePipe() {
   // CloseMessagePipe() above won't trigger connection error handler.
   // Explicitly call OnPipeConnectionError() so that associated endpoints will
   // get notified.
-  OnPipeConnectionError();
+  OnPipeConnectionError(true /* force_async_dispatch */);
 }
 
 void MultiplexRouter::PauseIncomingMethodCallProcessing() {
@@ -641,7 +642,7 @@ bool MultiplexRouter::OnPeerAssociatedEndpointClosed(
   return true;
 }
 
-void MultiplexRouter::OnPipeConnectionError() {
+void MultiplexRouter::OnPipeConnectionError(bool force_async_dispatch) {
   DCHECK_CALLED_ON_VALID_SEQUENCE(sequence_checker_);
 
   scoped_refptr<MultiplexRouter> protector(this);
@@ -664,10 +665,13 @@ void MultiplexRouter::OnPipeConnectionError() {
     UpdateEndpointStateMayRemove(endpoint.get(), PEER_ENDPOINT_CLOSED);
   }
 
-  ProcessTasks(connector_.during_sync_handle_watcher_callback()
-                   ? ALLOW_DIRECT_CLIENT_CALLS_FOR_SYNC_MESSAGES
-                   : ALLOW_DIRECT_CLIENT_CALLS,
-               connector_.task_runner());
+  ClientCallBehavior call_behavior = ALLOW_DIRECT_CLIENT_CALLS;
+  if (force_async_dispatch)
+    call_behavior = NO_DIRECT_CLIENT_CALLS;
+  else if (connector_.during_sync_handle_watcher_callback())
+    call_behavior = ALLOW_DIRECT_CLIENT_CALLS_FOR_SYNC_MESSAGES;
+
+  ProcessTasks(call_behavior, connector_.task_runner());
 }
 
 void MultiplexRouter::ProcessTasks(
@@ -879,7 +883,8 @@ void MultiplexRouter::MaybePostToProcessTasks(
   posted_to_process_tasks_ = true;
   posted_to_task_runner_ = task_runner;
   task_runner->PostTask(
-      FROM_HERE, base::Bind(&MultiplexRouter::LockAndCallProcessTasks, this));
+      FROM_HERE,
+      base::BindOnce(&MultiplexRouter::LockAndCallProcessTasks, this));
 }
 
 void MultiplexRouter::LockAndCallProcessTasks() {

mojo/public/cpp/bindings/lib/multiplex_router.h

@@ -173,7 +173,7 @@ class MOJO_CPP_BINDINGS_EXPORT MultiplexRouter
       InterfaceId id,
       const base::Optional<DisconnectReason>& reason) override;
 
-  void OnPipeConnectionError();
+  void OnPipeConnectionError(bool force_async_dispatch);
 
   // Specifies whether we are allowed to directly call into
   // InterfaceEndpointClient (given that we are already on the same sequence as

mojo/public/cpp/bindings/tests/associated_interface_unittest.cc

@@ -15,6 +15,7 @@
 #include "base/single_thread_task_runner.h"
 #include "base/synchronization/waitable_event.h"
 #include "base/task_scheduler/post_task.h"
+#include "base/test/bind_test_util.h"
 #include "base/test/scoped_task_environment.h"
 #include "base/threading/sequenced_task_runner_handle.h"
 #include "base/threading/thread.h"
@@ -1184,6 +1185,32 @@ TEST_F(AssociatedInterfaceTest, AssociateWithDisconnectedPipe) {
   sender->Send(42);
 }
 
+TEST_F(AssociatedInterfaceTest, AsyncErrorHandlersWhenClosingMasterInterface) {
+  // Ensures that associated interface error handlers are not invoked
+  // synchronously when the master interface pipe is closed. Regression test for
+  // https://crbug.com/864731.
+
+  IntegerSenderConnectionPtr connection_ptr;
+  IntegerSenderConnectionImpl connection(MakeRequest(&connection_ptr));
+
+  base::RunLoop loop;
+  bool error_handler_invoked = false;
+  IntegerSenderAssociatedPtr sender0;
+  connection_ptr->GetSender(MakeRequest(&sender0));
+  sender0.set_connection_error_handler(base::BindLambdaForTesting([&] {
+    error_handler_invoked = true;
+    loop.Quit();
+  }));
+
+  // This should not trigger the error handler synchronously...
+  connection_ptr.reset();
+  EXPECT_FALSE(error_handler_invoked);
+
+  // ...but it should be triggered once we spin the scheduler.
+  loop.Run();
+  EXPECT_TRUE(error_handler_invoked);
+}
+
 }  // namespace
 }  // namespace test
 }  // namespace mojo