Store per-frame task runners for IPC messages task execution

This CL stores per-frame task runners for IPC messages to be executed on
using routing_id. The task runners are stored on IPC::ChannelProxy::Context and will be
looked up each time when IPC messages arrives to the renderer.

Code-wise, this CL
- Introduces IPC::ChannelProxy::Context::{AddListenerTaskRunner, GetTaskRunner}
- {Associate, query} a task runner {to, from} a IPC routing_id
- Before this CL, all IPC tasks were PostTask to {ipc, listener} task runner, but after this CL,
 they are PostTask to the task runner associated to their IPC message's routing_id.
- Renames kInternalNavigation to kInternalNavigationAssociated,
- Associate frame routing_id to per-frame task runner, so the frame IPC messages are routed to
 the task runner.

Design-doc: https://docs.google.com/document/d/1dfQEPeeqOBigyzaeLR5dYaAgeeNDhCkWsPyxodD5xTA/edit?usp=sharing

Change-Id: Ie7500355ca489ee192f9de53f05d07774cf286fb
Reviewed-on: https://chromium-review.googlesource.com/c/chromium/src/+/1526067
Commit-Queue: Yuzu Saijo <yuzus@chromium.org>
Reviewed-by: Ken Rockot <rockot@google.com>
Reviewed-by: Kouhei Ueno <kouhei@chromium.org>
Reviewed-by: Daniel Cheng <dcheng@chromium.org>
Reviewed-by: Alexander Timin <altimin@chromium.org>
Reviewed-by: Kentaro Hara <haraken@chromium.org>
Cr-Commit-Position: refs/heads/master@{#662948}

content/renderer/navigation_client.cc

@@ -58,8 +58,8 @@ void NavigationClient::CommitFailedNavigation(
 
 void NavigationClient::Bind(mojom::NavigationClientAssociatedRequest request) {
   navigation_client_binding_.Bind(
-      std::move(request),
-      render_frame_->GetTaskRunner(blink::TaskType::kInternalNavigation));
+      std::move(request), render_frame_->GetTaskRunner(
+                              blink::TaskType::kInternalNavigationAssociated));
   SetDisconnectionHandler();
 }
 

content/renderer/render_thread_impl.cc

@@ -1069,6 +1069,10 @@ void RenderThreadImpl::AddRoute(int32_t routing_id, IPC::Listener* listener) {
   if (!frame)
     return;
 
+  GetChannel()->AddListenerTaskRunner(
+      routing_id,
+      frame->GetTaskRunner(blink::TaskType::kInternalNavigationAssociated));
+
   scoped_refptr<PendingFrameCreate> create(it->second);
   frame->BindFrame(it->second->browser_info(), it->second->TakeFrameRequest());
   pending_frame_creates_.erase(it);
@@ -1076,6 +1080,7 @@ void RenderThreadImpl::AddRoute(int32_t routing_id, IPC::Listener* listener) {
 
 void RenderThreadImpl::RemoveRoute(int32_t routing_id) {
   ChildThreadImpl::GetRouter()->RemoveRoute(routing_id);
+  GetChannel()->RemoveListenerTaskRunner(routing_id);
 }
 
 void RenderThreadImpl::RegisterPendingFrameCreate(

ipc/ipc_channel_proxy.cc

@@ -32,7 +32,7 @@ ChannelProxy::Context::Context(
     Listener* listener,
     const scoped_refptr<base::SingleThreadTaskRunner>& ipc_task_runner,
     const scoped_refptr<base::SingleThreadTaskRunner>& listener_task_runner)
-    : listener_task_runner_(listener_task_runner),
+    : default_listener_task_runner_(listener_task_runner),
       listener_(listener),
       ipc_task_runner_(ipc_task_runner),
       channel_connected_called_(false),
@@ -47,7 +47,8 @@ ChannelProxy::Context::Context(
   // Note, we currently make an exception for a NULL listener. That usage
   // basically works, but is outside the intent of ChannelProxy. This support
   // will disappear, so please don't rely on it. See crbug.com/364241
-  DCHECK(!listener || (ipc_task_runner_.get() != listener_task_runner_.get()));
+  DCHECK(!listener ||
+         (ipc_task_runner_.get() != default_listener_task_runner_.get()));
 }
 
 ChannelProxy::Context::~Context() = default;
@@ -85,9 +86,9 @@ bool ChannelProxy::Context::TryFilters(const Message& message) {
 
   if (message_filter_router_->TryFilters(message)) {
     if (message.dispatch_error()) {
-      listener_task_runner_->PostTask(
-          FROM_HERE,
-          base::BindOnce(&Context::OnDispatchBadMessage, this, message));
+      GetTaskRunner(message.routing_id())
+          ->PostTask(FROM_HERE, base::BindOnce(&Context::OnDispatchBadMessage,
+                                               this, message));
     }
 #if BUILDFLAG(IPC_MESSAGE_LOG_ENABLED)
     if (logger->Enabled())
@@ -126,8 +127,9 @@ bool ChannelProxy::Context::OnMessageReceived(const Message& message) {
 
 // Called on the IPC::Channel thread
 bool ChannelProxy::Context::OnMessageReceivedNoFilter(const Message& message) {
-  listener_task_runner_->PostTask(
-      FROM_HERE, base::BindOnce(&Context::OnDispatchMessage, this, message));
+  GetTaskRunner(message.routing_id())
+      ->PostTask(FROM_HERE,
+                 base::BindOnce(&Context::OnDispatchMessage, this, message));
   return true;
 }
 
@@ -145,8 +147,8 @@ void ChannelProxy::Context::OnChannelConnected(int32_t peer_pid) {
   // the filter is run on the IO thread.
   OnAddFilter();
 
-  // See above comment about using listener_task_runner_ here.
-  listener_task_runner_->PostTask(
+  // See above comment about using default_listener_task_runner_ here.
+  default_listener_task_runner_->PostTask(
       FROM_HERE, base::BindOnce(&Context::OnDispatchConnected, this));
 }
 
@@ -155,8 +157,8 @@ void ChannelProxy::Context::OnChannelError() {
   for (size_t i = 0; i < filters_.size(); ++i)
     filters_[i]->OnChannelError();
 
-  // See above comment about using listener_task_runner_ here.
-  listener_task_runner_->PostTask(
+  // See above comment about using default_listener_task_runner_ here.
+  default_listener_task_runner_->PostTask(
       FROM_HERE, base::BindOnce(&Context::OnDispatchError, this));
 }
 
@@ -164,7 +166,7 @@ void ChannelProxy::Context::OnChannelError() {
 void ChannelProxy::Context::OnAssociatedInterfaceRequest(
     const std::string& interface_name,
     mojo::ScopedInterfaceEndpointHandle handle) {
-  listener_task_runner_->PostTask(
+  default_listener_task_runner_->PostTask(
       FROM_HERE, base::BindOnce(&Context::OnDispatchAssociatedInterfaceRequest,
                                 this, interface_name, std::move(handle)));
 }
@@ -331,6 +333,40 @@ void ChannelProxy::Context::OnDispatchMessage(const Message& message) {
 #endif
 }
 
+// Called on the listener's thread.
+void ChannelProxy::Context::AddListenerTaskRunner(
+    int32_t routing_id,
+    scoped_refptr<base::SingleThreadTaskRunner> task_runner) {
+  DCHECK(default_listener_task_runner_->BelongsToCurrentThread());
+  DCHECK(task_runner);
+  base::AutoLock lock(listener_thread_task_runners_lock_);
+  if (!base::ContainsKey(listener_thread_task_runners_, routing_id))
+    listener_thread_task_runners_.insert({routing_id, std::move(task_runner)});
+}
+
+// Called on the listener's thread.
+void ChannelProxy::Context::RemoveListenerTaskRunner(int32_t routing_id) {
+  DCHECK(default_listener_task_runner_->BelongsToCurrentThread());
+  base::AutoLock lock(listener_thread_task_runners_lock_);
+  if (base::ContainsKey(listener_thread_task_runners_, routing_id))
+    listener_thread_task_runners_.erase(routing_id);
+}
+
+// Called on the IPC::Channel thread.
+base::SingleThreadTaskRunner* ChannelProxy::Context::GetTaskRunner(
+    int32_t routing_id) {
+  DCHECK(ipc_task_runner_->BelongsToCurrentThread());
+  if (routing_id == MSG_ROUTING_NONE)
+    return default_listener_task_runner_.get();
+
+  base::AutoLock lock(listener_thread_task_runners_lock_);
+  base::SingleThreadTaskRunner* task_runner =
+      listener_thread_task_runners_[routing_id].get();
+  if (task_runner)
+    return task_runner;
+  return default_listener_task_runner_.get();
+}
+
 // Called on the listener's thread
 void ChannelProxy::Context::OnDispatchConnected() {
   if (channel_connected_called_)

ipc/ipc_channel_proxy.h

@@ -263,7 +263,7 @@ class COMPONENT_EXPORT(IPC) ChannelProxy : public Sender {
     }
 
     scoped_refptr<base::SingleThreadTaskRunner> listener_task_runner() {
-      return listener_task_runner_;
+      return default_listener_task_runner_;
     }
 
     // Dispatches a message on the listener thread.
@@ -272,6 +272,18 @@ class COMPONENT_EXPORT(IPC) ChannelProxy : public Sender {
     // Sends |message| from appropriate thread.
     void Send(Message* message, const char* debug_name);
 
+    // Adds |task_runner| for the task to be executed later.
+    void AddListenerTaskRunner(
+        int32_t routing_id,
+        scoped_refptr<base::SingleThreadTaskRunner> task_runner);
+
+    // Removes task runner for |routing_id|.
+    void RemoveListenerTaskRunner(int32_t routing_id);
+
+    // Called on the IPC::Channel thread.
+    // Returns the task runner associated with |routing_id|.
+    base::SingleThreadTaskRunner* GetTaskRunner(int32_t routing_id);
+
    protected:
     friend class base::RefCountedThreadSafe<Context>;
     ~Context() override;
@@ -336,7 +348,13 @@ class COMPONENT_EXPORT(IPC) ChannelProxy : public Sender {
         const std::string& name,
         const GenericAssociatedInterfaceFactory& factory);
 
-    scoped_refptr<base::SingleThreadTaskRunner> listener_task_runner_;
+    base::Lock listener_thread_task_runners_lock_;
+    // Map of routing_id and listener's thread task runner.
+    std::map<int32_t, scoped_refptr<base::SingleThreadTaskRunner>>
+        listener_thread_task_runners_
+            GUARDED_BY(listener_thread_task_runners_lock_);
+
+    scoped_refptr<base::SingleThreadTaskRunner> default_listener_task_runner_;
     Listener* listener_;
 
     // List of filters.  This is only accessed on the IPC thread.

ipc/ipc_sync_channel.cc

@@ -578,6 +578,16 @@ SyncChannel::SyncChannel(
   StartWatching();
 }
 
+void SyncChannel::AddListenerTaskRunner(
+    int32_t routing_id,
+    scoped_refptr<base::SingleThreadTaskRunner> task_runner) {
+  context()->AddListenerTaskRunner(routing_id, std::move(task_runner));
+}
+
+void SyncChannel::RemoveListenerTaskRunner(int32_t routing_id) {
+  context()->RemoveListenerTaskRunner(routing_id);
+}
+
 SyncChannel::~SyncChannel() = default;
 
 void SyncChannel::SetRestrictDispatchChannelGroup(int group) {

ipc/ipc_sync_channel.h

@@ -96,6 +96,12 @@ class COMPONENT_EXPORT(IPC) SyncChannel : public ChannelProxy {
       const scoped_refptr<base::SingleThreadTaskRunner>& listener_task_runner,
       base::WaitableEvent* shutdown_event);
 
+  void AddListenerTaskRunner(
+      int32_t routing_id,
+      scoped_refptr<base::SingleThreadTaskRunner> task_runner);
+
+  void RemoveListenerTaskRunner(int32_t routing_id);
+
   ~SyncChannel() override;
 
   bool Send(Message* message) override;

third_party/blink/public/platform/task_type.h

@@ -203,8 +203,12 @@ enum class TaskType : unsigned char {
   // Task used for ContentCapture.
   kInternalContentCapture = 61,
 
-  // Task used for Navigations.
-  kInternalNavigation = 63,
+  // Navigation tasks and tasks which have to run in order with them, including
+  // legacy IPCs and channel associated interfaces.
+  // Note that the ordering between tasks related to different frames is not
+  // always guaranteed - tasks belonging to different frames can be reordered
+  // when one of the frames is frozen.
+  kInternalNavigationAssociated = 63,
 
   ///////////////////////////////////////
   // The following task types are only for thread-local queues.

third_party/blink/renderer/platform/scheduler/main_thread/frame_scheduler_impl.cc

@@ -470,7 +470,7 @@ base::Optional<QueueTraits> FrameSchedulerImpl::CreateQueueTraitsForTaskType(
     case TaskType::kInternalTranslation:
       return ForegroundOnlyTaskQueueTraits();
     // Navigation IPCs do not run using virtual time to avoid hanging.
-    case TaskType::kInternalNavigation:
+    case TaskType::kInternalNavigationAssociated:
       return DoesNotUseVirtualTimeTaskQueueTraits();
     case TaskType::kDeprecatedNone:
     case TaskType::kMainThreadTaskQueueV8:

third_party/blink/renderer/platform/scheduler/main_thread/task_type_names.cc

@@ -129,8 +129,8 @@ const char* TaskTypeNames::TaskTypeToString(TaskType task_type) {
       return "InternalTranslation";
     case TaskType::kInternalContentCapture:
       return "InternalContentCapture";
-    case TaskType::kInternalNavigation:
-      return "InternalNavigation";
+    case TaskType::kInternalNavigationAssociated:
+      return "InternalNavigationAssociated";
     case TaskType::kCount:
       return "Count";
   }

third_party/blink/renderer/platform/scheduler/worker/worker_scheduler.cc

@@ -165,7 +165,7 @@ scoped_refptr<base::SingleThreadTaskRunner> WorkerScheduler::GetTaskRunner(
     case TaskType::kInternalIPC:
     case TaskType::kInternalInspector:
     case TaskType::kInternalTest:
-    case TaskType::kInternalNavigation:
+    case TaskType::kInternalNavigationAssociated:
       // UnthrottledTaskRunner is generally discouraged in future.
       // TODO(nhiroki): Identify which tasks can be throttled / suspendable and
       // move them into other task runners. See also comments in