Make IPC listener task runner changeable

This CL adds an argument for listener task runner to
IPC::ChannelProxy::Create and IPC::SyncChannel::Create so that we
can have more flexible task scheduling. Now all arguments are
base::ThreadTaskRunnerHandle::Get() and this CL doesn't change the
behavior. This CL is just a preparation.

We've already known that channel proxy's OnMessageReceivedNoFilter
posts the most heaviest tasks to the listener task runner.

Doc: https://docs.google.com/document/d/1yOhE6-1HLb3aeNWmoa9O2LlnegWjd4awYUn2OhUL4vk/edit#

Bug: 780785
Cq-Include-Trybots: master.tryserver.chromium.android:android_optional_gpu_tests_rel;master.tryserver.chromium.linux:linux_optional_gpu_tests_rel;master.tryserver.chromium.mac:mac_optional_gpu_tests_rel;master.tryserver.chromium.win:win_optional_gpu_tests_rel
Change-Id: Ia0d798c55aeb739d7150e9368350ecd28f4159c6
Reviewed-on: https://chromium-review.googlesource.com/750922
Reviewed-by: John Abd-El-Malek <[email protected]>
Reviewed-by: Ken Rockot <[email protected]>
Reviewed-by: Alexander Timin <[email protected]>
Commit-Queue: Hajime Hoshi <[email protected]>
Cr-Commit-Position: refs/heads/master@{#515115}

chrome/app_shim/chrome_main_app_mode_mac.mm

@@ -228,7 +228,8 @@ bool SendFocusApp(apps::AppShimFocusType focus_type,
               mojo::edk::CreateClientHandle(
                   mojo::edk::NamedPlatformHandle(socket_path.value())))),
           g_io_thread->task_runner().get()),
-      this, g_io_thread->task_runner().get());
+      this, g_io_thread->task_runner().get(),
+      base::ThreadTaskRunnerHandle::Get());
 
   bool launched_by_chrome = base::CommandLine::ForCurrentProcess()->HasSwitch(
       app_mode::kLaunchedByChromeProcessId);

chrome/browser/apps/app_shim/app_shim_host_mac.cc

@@ -37,7 +37,8 @@ void AppShimHost::ServeChannel(mojo::edk::ScopedPlatformHandle handle) {
               .get()),
       this,
       content::BrowserThread::GetTaskRunnerForThread(content::BrowserThread::IO)
-          .get());
+          .get(),
+      base::ThreadTaskRunnerHandle::Get());
 }
 
 base::FilePath AppShimHost::GetProfilePath() const {

chrome/browser/apps/app_shim/app_shim_host_manager_browsertest_mac.mm

@@ -80,7 +80,8 @@ void Send(const T& message) {
               mojo::edk::CreateClientHandle(
                   mojo::edk::NamedPlatformHandle(socket_path.value())))),
           io_thread_.task_runner().get()),
-      this, io_thread_.task_runner().get());
+      this, io_thread_.task_runner().get(),
+      base::ThreadTaskRunnerHandle::Get());
 }
 
 TestShimClient::~TestShimClient() {

chrome/browser/printing/cloud_print/test/cloud_print_proxy_process_browsertest.cc

@@ -258,7 +258,8 @@ int CloudPrintMockService_Main(SetExpectationsCallback set_expectations) {
                       mojo::edk::NamedPlatformHandle(startup_channel_name))))
               .release(),
           IPC::Channel::MODE_CLIENT, &listener,
-          service_process.io_task_runner());
+          service_process.io_task_runner(),
+          base::ThreadTaskRunnerHandle::Get());
 
   base::RunLoop().Run();
   if (!Mock::VerifyAndClearExpectations(&server))
@@ -422,7 +423,8 @@ base::Process CloudPrintProxyPolicyStartupTest::Launch(
               mojo::edk::TransportProtocol::kLegacy,
               mojo::edk::CreateServerHandle(startup_channel_handle_)))
           .release(),
-      IPC::Channel::MODE_SERVER, this, IOTaskRunner());
+      IPC::Channel::MODE_SERVER, this, IOTaskRunner(),
+      base::ThreadTaskRunnerHandle::Get());
 
   base::Process process = SpawnChild(name);
   EXPECT_TRUE(process.IsValid());

components/nacl/browser/nacl_process_host.cc

@@ -900,6 +900,7 @@ bool NaClProcessHost::StartPPAPIProxy(
 
   ipc_proxy_channel_ = IPC::ChannelProxy::Create(
       channel_handle.release(), IPC::Channel::MODE_CLIENT, NULL,
+      base::ThreadTaskRunnerHandle::Get().get(),
       base::ThreadTaskRunnerHandle::Get().get());
   // Create the browser ppapi host and enable PPAPI message dispatching to the
   // browser process.

components/nacl/loader/nacl_listener.cc

@@ -222,6 +222,7 @@ class FileTokenMessageFilter : public IPC::MessageFilter {
 
 void NaClListener::Listen() {
   channel_ = IPC::SyncChannel::Create(this, io_thread_.task_runner().get(),
+                                      base::ThreadTaskRunnerHandle::Get(),
                                       &shutdown_event_);
   filter_ = channel_->CreateSyncMessageFilter();
   channel_->AddFilter(new FileTokenMessageFilter());

components/nacl/loader/nonsfi/nonsfi_listener.cc

@@ -53,12 +53,12 @@ void NonSfiListener::Listen() {
   mojo::ScopedMessagePipeHandle channel_handle;
   std::unique_ptr<service_manager::ServiceContext> service_context =
       CreateNaClServiceContext(io_thread_.task_runner(), &channel_handle);
-  channel_ = IPC::SyncChannel::Create(channel_handle.release(),
-                                      IPC::Channel::MODE_CLIENT,
-                                      this,  // As a Listener.
-                                      io_thread_.task_runner(),
-                                      true,  // Create pipe now.
-                                      &shutdown_event_);
+  channel_ = IPC::SyncChannel::Create(
+      channel_handle.release(), IPC::Channel::MODE_CLIENT,
+      this,  // As a Listener.
+      io_thread_.task_runner(), base::ThreadTaskRunnerHandle::Get(),
+      true,  // Create pipe now.
+      &shutdown_event_);
   base::RunLoop().Run();
 }
 

components/nacl/renderer/manifest_service_channel.cc

@@ -33,6 +33,7 @@ ManifestServiceChannel::ManifestServiceChannel(
           IPC::Channel::MODE_CLIENT,
           this,
           content::RenderThread::Get()->GetIOTaskRunner(),
+          base::ThreadTaskRunnerHandle::Get(),
           true,
           waitable_event)),
       peer_pid_(base::kNullProcessId),

components/nacl/renderer/ppb_nacl_private_impl.cc

@@ -525,7 +525,8 @@ void PPBNaClPrivate::LaunchSelLdr(
       *translator_channel = IPC::SyncChannel::Create(
           instance_info.channel_handle, IPC::Channel::MODE_CLIENT,
           new NoOpListener, content::RenderThread::Get()->GetIOTaskRunner(),
-          true, content::RenderThread::Get()->GetShutdownEvent());
+          base::ThreadTaskRunnerHandle::Get(), true,
+          content::RenderThread::Get()->GetShutdownEvent());
     } else {
       // Save the channel handle for when StartPpapiProxy() is called.
       NaClPluginInstance* nacl_plugin_instance =

content/browser/browser_associated_interface_unittest.cc

@@ -52,8 +52,9 @@ class ProxyRunner : public IPC::Listener {
       factory = IPC::ChannelMojo::CreateClientFactory(
           std::move(pipe), ipc_task_runner);
     }
-    channel_ = IPC::ChannelProxy::Create(
-        std::move(factory), this, ipc_task_runner);
+    channel_ =
+        IPC::ChannelProxy::Create(std::move(factory), this, ipc_task_runner,
+                                  base::ThreadTaskRunnerHandle::Get());
   }
 
   void ShutDown() { channel_.reset(); }

content/browser/renderer_host/render_process_host_impl.cc

@@ -1604,12 +1604,14 @@ void RenderProcessHostImpl::InitializeChannelProxy() {
   // See crbug.com/526842 for details.
 #if defined(OS_ANDROID)
   if (GetContentClient()->UsingSynchronousCompositing()) {
-    channel_ = IPC::SyncChannel::Create(
-        this, io_task_runner.get(), &never_signaled_);
+    channel_ = IPC::SyncChannel::Create(this, io_task_runner.get(),
+                                        base::ThreadTaskRunnerHandle::Get(),
+                                        &never_signaled_);
   }
 #endif  // OS_ANDROID
   if (!channel_)
-    channel_.reset(new IPC::ChannelProxy(this, io_task_runner.get()));
+    channel_.reset(new IPC::ChannelProxy(this, io_task_runner.get(),
+                                         base::ThreadTaskRunnerHandle::Get()));
   channel_->Init(std::move(channel_factory), true /* create_pipe_now */);
 
   // Note that Channel send is effectively paused and unpaused at various points

content/child/child_thread_impl.cc

@@ -434,6 +434,7 @@ void ChildThreadImpl::Init(const Options& options) {
 
   channel_ =
       IPC::SyncChannel::Create(this, ChildProcess::current()->io_task_runner(),
+                               base::ThreadTaskRunnerHandle::Get(),
                                ChildProcess::current()->GetShutDownEvent());
 #if BUILDFLAG(IPC_MESSAGE_LOG_ENABLED)
   if (!IsInBrowserProcess())

content/renderer/render_thread_impl_browsertest.cc

@@ -197,10 +197,10 @@ class RenderThreadImplBrowserTest : public testing::Test {
     child_connection_->BindInterface(IPC::mojom::ChannelBootstrap::Name_,
                                      std::move(pipe.handle1));
 
-    channel_ =
-        IPC::ChannelProxy::Create(IPC::ChannelMojo::CreateServerFactory(
-                                      std::move(pipe.handle0), io_task_runner),
-                                  nullptr, io_task_runner);
+    channel_ = IPC::ChannelProxy::Create(
+        IPC::ChannelMojo::CreateServerFactory(std::move(pipe.handle0),
+                                              io_task_runner),
+        nullptr, io_task_runner, base::ThreadTaskRunnerHandle::Get());
 
     mock_process_.reset(new MockRenderProcess);
     test_task_counter_ = base::MakeRefCounted<TestTaskCounter>();

gpu/ipc/client/gpu_channel_host.cc

@@ -68,9 +68,9 @@ void GpuChannelHost::Connect(const IPC::ChannelHandle& channel_handle,
   // here since we need to filter everything to route it to the right thread.
   scoped_refptr<base::SingleThreadTaskRunner> io_task_runner =
       factory_->GetIOThreadTaskRunner();
-  channel_ = IPC::SyncChannel::Create(channel_handle, IPC::Channel::MODE_CLIENT,
-                                      nullptr, io_task_runner.get(), true,
-                                      shutdown_event);
+  channel_ = IPC::SyncChannel::Create(
+      channel_handle, IPC::Channel::MODE_CLIENT, nullptr, io_task_runner.get(),
+      base::ThreadTaskRunnerHandle::Get(), true, shutdown_event);
 
   channel_filter_ = new MessageFilter();
 

gpu/ipc/service/gpu_channel.cc

@@ -318,6 +318,7 @@ SyncChannelFilteredSender::SyncChannelFilteredSender(
                                         IPC::Channel::MODE_SERVER,
                                         listener,
                                         ipc_task_runner,
+                                        base::ThreadTaskRunnerHandle::Get(),
                                         false,
                                         shutdown_event)) {}
 

ipc/ipc_channel_mojo_unittest.cc

@@ -657,8 +657,9 @@ class ChannelProxyRunner {
   void CreateProxy(IPC::Listener* listener) {
     io_thread_.StartWithOptions(
         base::Thread::Options(base::MessageLoop::TYPE_IO, 0));
-    proxy_ = IPC::SyncChannel::Create(
-        listener, io_thread_.task_runner(), &never_signaled_);
+    proxy_ = IPC::SyncChannel::Create(listener, io_thread_.task_runner(),
+                                      base::ThreadTaskRunnerHandle::Get(),
+                                      &never_signaled_);
   }
 
   void RunProxy() {

ipc/ipc_channel_proxy.cc

@@ -30,8 +30,9 @@ namespace IPC {
 
 ChannelProxy::Context::Context(
     Listener* listener,
-    const scoped_refptr<base::SingleThreadTaskRunner>& ipc_task_runner)
-    : listener_task_runner_(base::ThreadTaskRunnerHandle::Get()),
+    const scoped_refptr<base::SingleThreadTaskRunner>& ipc_task_runner,
+    const scoped_refptr<base::SingleThreadTaskRunner>& listener_task_runner)
+    : listener_task_runner_(listener_task_runner),
       listener_(listener),
       ipc_task_runner_(ipc_task_runner),
       channel_connected_called_(false),
@@ -395,9 +396,10 @@ std::unique_ptr<ChannelProxy> ChannelProxy::Create(
     const IPC::ChannelHandle& channel_handle,
     Channel::Mode mode,
     Listener* listener,
-    const scoped_refptr<base::SingleThreadTaskRunner>& ipc_task_runner) {
+    const scoped_refptr<base::SingleThreadTaskRunner>& ipc_task_runner,
+    const scoped_refptr<base::SingleThreadTaskRunner>& listener_task_runner) {
   std::unique_ptr<ChannelProxy> channel(
-      new ChannelProxy(listener, ipc_task_runner));
+      new ChannelProxy(listener, ipc_task_runner, listener_task_runner));
   channel->Init(channel_handle, mode, true);
   return channel;
 }
@@ -406,9 +408,10 @@ std::unique_ptr<ChannelProxy> ChannelProxy::Create(
 std::unique_ptr<ChannelProxy> ChannelProxy::Create(
     std::unique_ptr<ChannelFactory> factory,
     Listener* listener,
-    const scoped_refptr<base::SingleThreadTaskRunner>& ipc_task_runner) {
+    const scoped_refptr<base::SingleThreadTaskRunner>& ipc_task_runner,
+    const scoped_refptr<base::SingleThreadTaskRunner>& listener_task_runner) {
   std::unique_ptr<ChannelProxy> channel(
-      new ChannelProxy(listener, ipc_task_runner));
+      new ChannelProxy(listener, ipc_task_runner, listener_task_runner));
   channel->Init(std::move(factory), true);
   return channel;
 }
@@ -422,8 +425,10 @@ ChannelProxy::ChannelProxy(Context* context)
 
 ChannelProxy::ChannelProxy(
     Listener* listener,
-    const scoped_refptr<base::SingleThreadTaskRunner>& ipc_task_runner)
-    : context_(new Context(listener, ipc_task_runner)), did_init_(false) {
+    const scoped_refptr<base::SingleThreadTaskRunner>& ipc_task_runner,
+    const scoped_refptr<base::SingleThreadTaskRunner>& listener_task_runner)
+    : context_(new Context(listener, ipc_task_runner, listener_task_runner)),
+      did_init_(false) {
 #if defined(ENABLE_IPC_FUZZER)
   outgoing_message_filter_ = NULL;
 #endif

ipc/ipc_channel_proxy.h

@@ -95,17 +95,20 @@ class IPC_EXPORT ChannelProxy : public Sender {
       const IPC::ChannelHandle& channel_handle,
       Channel::Mode mode,
       Listener* listener,
-      const scoped_refptr<base::SingleThreadTaskRunner>& ipc_task_runner);
+      const scoped_refptr<base::SingleThreadTaskRunner>& ipc_task_runner,
+      const scoped_refptr<base::SingleThreadTaskRunner>& listener_task_runner);
 
   static std::unique_ptr<ChannelProxy> Create(
       std::unique_ptr<ChannelFactory> factory,
       Listener* listener,
-      const scoped_refptr<base::SingleThreadTaskRunner>& ipc_task_runner);
+      const scoped_refptr<base::SingleThreadTaskRunner>& ipc_task_runner,
+      const scoped_refptr<base::SingleThreadTaskRunner>& listener_task_runner);
 
   // Constructs a ChannelProxy without initializing it.
   ChannelProxy(
       Listener* listener,
-      const scoped_refptr<base::SingleThreadTaskRunner>& ipc_task_runner);
+      const scoped_refptr<base::SingleThreadTaskRunner>& ipc_task_runner,
+      const scoped_refptr<base::SingleThreadTaskRunner>& listener_task_runner);
 
   ~ChannelProxy() override;
 
@@ -244,7 +247,9 @@ class IPC_EXPORT ChannelProxy : public Sender {
                   public Listener {
    public:
     Context(Listener* listener,
-            const scoped_refptr<base::SingleThreadTaskRunner>& ipc_thread);
+            const scoped_refptr<base::SingleThreadTaskRunner>& ipc_task_runner,
+            const scoped_refptr<base::SingleThreadTaskRunner>&
+                listener_task_runner);
     void ClearIPCTaskRunner();
     base::SingleThreadTaskRunner* ipc_task_runner() const {
       return ipc_task_runner_.get();

ipc/ipc_channel_proxy_unittest.cc

@@ -247,7 +247,7 @@ class IPCChannelProxyTest : public IPCChannelMojoTestBase {
     listener_.reset(new QuitListener());
     channel_proxy_ = IPC::ChannelProxy::Create(
         TakeHandle().release(), IPC::Channel::MODE_SERVER, listener_.get(),
-        thread_->task_runner());
+        thread_->task_runner(), base::ThreadTaskRunnerHandle::Get());
   }
 
   void TearDown() override {

ipc/ipc_cpu_perftest.cc

@@ -225,11 +225,12 @@ class ChannelSteadyPingPongTest : public IPCChannelMojoTestBase {
           base::WaitableEvent::InitialState::NOT_SIGNALED);
       channel_proxy = IPC::SyncChannel::Create(
           TakeHandle().release(), IPC::Channel::MODE_SERVER, &listener,
-          GetIOThreadTaskRunner(), false, shutdown_event.get());
+          GetIOThreadTaskRunner(), base::ThreadTaskRunnerHandle::Get(), false,
+          shutdown_event.get());
     } else {
       channel_proxy = IPC::ChannelProxy::Create(
           TakeHandle().release(), IPC::Channel::MODE_SERVER, &listener,
-          GetIOThreadTaskRunner());
+          GetIOThreadTaskRunner(), base::ThreadTaskRunnerHandle::Get());
     }
     listener.Init(channel_proxy.get());
 

ipc/ipc_mojo_perftest.cc

@@ -189,7 +189,7 @@ class MojoChannelPerfTest : public IPCChannelMojoTestBase {
     PerformanceChannelListener listener("ChannelProxy");
     auto channel_proxy = IPC::ChannelProxy::Create(
         TakeHandle().release(), IPC::Channel::MODE_SERVER, &listener,
-        GetIOThreadTaskRunner());
+        GetIOThreadTaskRunner(), base::ThreadTaskRunnerHandle::Get());
     listener.Init(channel_proxy.get());
 
     LockThreadAffinity thread_locker(kSharedCore);
@@ -222,7 +222,8 @@ class MojoChannelPerfTest : public IPCChannelMojoTestBase {
         base::WaitableEvent::InitialState::NOT_SIGNALED);
     auto channel_proxy = IPC::SyncChannel::Create(
         TakeHandle().release(), IPC::Channel::MODE_SERVER, &listener,
-        GetIOThreadTaskRunner(), false, &shutdown_event);
+        GetIOThreadTaskRunner(), base::ThreadTaskRunnerHandle::Get(), false,
+        &shutdown_event);
     listener.Init(channel_proxy.get());
 
     LockThreadAffinity thread_locker(kSharedCore);

ipc/ipc_perftest_util.cc

@@ -112,9 +112,9 @@ int MojoPerfTestClient::Run(MojoHandle handle) {
   LockThreadAffinity thread_locker(kSharedCore);
 
   base::RunLoop run_loop;
-  std::unique_ptr<ChannelProxy> channel =
-      IPC::ChannelProxy::Create(handle_.release(), Channel::MODE_CLIENT,
-                                listener_.get(), GetIOThreadTaskRunner());
+  std::unique_ptr<ChannelProxy> channel = IPC::ChannelProxy::Create(
+      handle_.release(), Channel::MODE_CLIENT, listener_.get(),
+      GetIOThreadTaskRunner(), base::ThreadTaskRunnerHandle::Get());
   listener_->Init(channel.get(), run_loop.QuitWhenIdleClosure());
   run_loop.Run();
   return 0;

ipc/ipc_sync_channel.cc

@@ -350,12 +350,12 @@ base::LazyInstance<base::ThreadLocalPointer<
 SyncChannel::SyncContext::SyncContext(
     Listener* listener,
     const scoped_refptr<base::SingleThreadTaskRunner>& ipc_task_runner,
+    const scoped_refptr<base::SingleThreadTaskRunner>& listener_task_runner,
     WaitableEvent* shutdown_event)
-    : ChannelProxy::Context(listener, ipc_task_runner),
+    : ChannelProxy::Context(listener, ipc_task_runner, listener_task_runner),
       received_sync_msgs_(ReceivedSyncMsgQueue::AddContext()),
       shutdown_event_(shutdown_event),
-      restrict_dispatch_group_(kRestrictDispatchGroup_None) {
-}
+      restrict_dispatch_group_(kRestrictDispatchGroup_None) {}
 
 void SyncChannel::SyncContext::OnSendDoneEventSignaled(
     base::RunLoop* nested_loop,
@@ -529,10 +529,11 @@ std::unique_ptr<SyncChannel> SyncChannel::Create(
     Channel::Mode mode,
     Listener* listener,
     const scoped_refptr<base::SingleThreadTaskRunner>& ipc_task_runner,
+    const scoped_refptr<base::SingleThreadTaskRunner>& listener_task_runner,
     bool create_pipe_now,
     base::WaitableEvent* shutdown_event) {
   std::unique_ptr<SyncChannel> channel =
-      Create(listener, ipc_task_runner, shutdown_event);
+      Create(listener, ipc_task_runner, listener_task_runner, shutdown_event);
   channel->Init(channel_handle, mode, create_pipe_now);
   return channel;
 }
@@ -542,10 +543,11 @@ std::unique_ptr<SyncChannel> SyncChannel::Create(
     std::unique_ptr<ChannelFactory> factory,
     Listener* listener,
     const scoped_refptr<base::SingleThreadTaskRunner>& ipc_task_runner,
+    const scoped_refptr<base::SingleThreadTaskRunner>& listener_task_runner,
     bool create_pipe_now,
     base::WaitableEvent* shutdown_event) {
   std::unique_ptr<SyncChannel> channel =
-      Create(listener, ipc_task_runner, shutdown_event);
+      Create(listener, ipc_task_runner, listener_task_runner, shutdown_event);
   channel->Init(std::move(factory), create_pipe_now);
   return channel;
 }
@@ -554,16 +556,21 @@ std::unique_ptr<SyncChannel> SyncChannel::Create(
 std::unique_ptr<SyncChannel> SyncChannel::Create(
     Listener* listener,
     const scoped_refptr<base::SingleThreadTaskRunner>& ipc_task_runner,
+    const scoped_refptr<base::SingleThreadTaskRunner>& listener_task_runner,
     WaitableEvent* shutdown_event) {
-  return base::WrapUnique(
-      new SyncChannel(listener, ipc_task_runner, shutdown_event));
+  return base::WrapUnique(new SyncChannel(
+      listener, ipc_task_runner, listener_task_runner, shutdown_event));
 }
 
 SyncChannel::SyncChannel(
     Listener* listener,
     const scoped_refptr<base::SingleThreadTaskRunner>& ipc_task_runner,
+    const scoped_refptr<base::SingleThreadTaskRunner>& listener_task_runner,
     WaitableEvent* shutdown_event)
-    : ChannelProxy(new SyncContext(listener, ipc_task_runner, shutdown_event)),
+    : ChannelProxy(new SyncContext(listener,
+                                   ipc_task_runner,
+                                   listener_task_runner,
+                                   shutdown_event)),
       sync_handle_registry_(mojo::SyncHandleRegistry::current()) {
   // The current (listener) thread must be distinct from the IPC thread, or else
   // sending synchronous messages will deadlock.