gpu: Move out of order / control message handling to message filter.

gpu/ipc/service/gpu_channel.cc

 // Copyright (c) 2012 The Chromium Authors. All rights reserved.
 // Use of this source code is governed by a BSD-style license that can be
 // found in the LICENSE file.
 
 #include "gpu/ipc/service/gpu_channel.h"
 
 #include <utility>
 
 #if defined(OS_WIN)
 #include <windows.h>
@@ skipping 53 matching lines @@
 // below this threshold.
 const int64_t kStopPreemptThresholdMs = kVsyncIntervalMs;
 
 CommandBufferId GenerateCommandBufferId(int channel_id, int32_t route_id) {
   return CommandBufferId::FromUnsafeValue(
       (static_cast<uint64_t>(channel_id) << 32) | route_id);
 }
 
 }  // anonymous namespace
 
-scoped_refptr<GpuChannelMessageQueue> GpuChannelMessageQueue::Create(
-    GpuChannel* channel,
-    scoped_refptr<base::SingleThreadTaskRunner> io_task_runner,
-    scoped_refptr<PreemptionFlag> preempting_flag,
-    scoped_refptr<PreemptionFlag> preempted_flag,
-    SyncPointManager* sync_point_manager) {
-  return new GpuChannelMessageQueue(
-      channel, std::move(io_task_runner), std::move(preempting_flag),
-      std::move(preempted_flag), sync_point_manager);
-}
-
 GpuChannelMessageQueue::GpuChannelMessageQueue(
     GpuChannel* channel,
+    scoped_refptr<base::SingleThreadTaskRunner> main_task_runner,
     scoped_refptr<base::SingleThreadTaskRunner> io_task_runner,
     scoped_refptr<PreemptionFlag> preempting_flag,
     scoped_refptr<PreemptionFlag> preempted_flag,
     SyncPointManager* sync_point_manager)
-    : enabled_(true),
-      scheduled_(true),
-      channel_(channel),
-      preemption_state_(IDLE),
+    : channel_(channel),
       max_preemption_time_(
           base::TimeDelta::FromMilliseconds(kMaxPreemptTimeMs)),
       timer_(new base::OneShotTimer),
       sync_point_order_data_(sync_point_manager->CreateSyncPointOrderData()),
+      main_task_runner_(std::move(main_task_runner)),
       io_task_runner_(std::move(io_task_runner)),
       preempting_flag_(std::move(preempting_flag)),
       preempted_flag_(std::move(preempted_flag)),
       sync_point_manager_(sync_point_manager) {
   timer_->SetTaskRunner(io_task_runner_);
   io_thread_checker_.DetachFromThread();
 }
 
 GpuChannelMessageQueue::~GpuChannelMessageQueue() {
-  DCHECK(!enabled_);
   DCHECK(channel_messages_.empty());
 }
 
-void GpuChannelMessageQueue::Disable() {
-  {
-    base::AutoLock auto_lock(channel_lock_);
-    DCHECK(enabled_);
-    enabled_ = false;
-  }
-
-  // We guarantee that the queues will no longer be modified after enabled_
-  // is set to false, it is now safe to modify the queue without the lock.
-  // All public facing modifying functions check enabled_ while all
-  // private modifying functions DCHECK(enabled_) to enforce this.
+void GpuChannelMessageQueue::Destroy() {
+  // We guarantee that the queue will no longer be modified after Destroy is
+  // called, it is now safe to modify the queue without the lock. All public
+  // facing modifying functions check enabled_ while all private modifying
+  // functions DCHECK(enabled_) to enforce this.
   while (!channel_messages_.empty()) {
     const IPC::Message& msg = channel_messages_.front()->message;
     if (msg.is_sync()) {
       IPC::Message* reply = IPC::SyncMessage::GenerateReply(&msg);
       reply->set_reply_error();
       channel_->Send(reply);
     }
     channel_messages_.pop_front();
   }
 
-  if (sync_point_order_data_) {
-    sync_point_order_data_->Destroy();
-    sync_point_order_data_ = nullptr;
-  }
+  sync_point_order_data_->Destroy();
 
+  // Destroy timer on io thread.
   io_task_runner_->PostTask(
-      FROM_HERE, base::Bind(&GpuChannelMessageQueue::DisableIO, this));
-}
+      FROM_HERE, base::Bind([](std::unique_ptr<base::OneShotTimer>) {},
+                            base::Passed(&timer_)));
 
-void GpuChannelMessageQueue::DisableIO() {
-  DCHECK(io_thread_checker_.CalledOnValidThread());
-  timer_ = nullptr;
+  channel_ = nullptr;
 }
 
 bool GpuChannelMessageQueue::IsScheduled() const {
   base::AutoLock lock(channel_lock_);
   return scheduled_;
 }
 
 void GpuChannelMessageQueue::SetScheduled(bool scheduled) {
   base::AutoLock lock(channel_lock_);
-  DCHECK(enabled_);
   if (scheduled_ == scheduled)
     return;
   scheduled_ = scheduled;
   if (scheduled)
-    channel_->PostHandleMessage();
+    PostHandleMessageOnQueue();
   if (preempting_flag_) {
     io_task_runner_->PostTask(
         FROM_HERE,
         base::Bind(&GpuChannelMessageQueue::UpdatePreemptionState, this));
   }
 }
 
-bool GpuChannelMessageQueue::PushBackMessage(const IPC::Message& message) {
+void GpuChannelMessageQueue::PushBackMessage(const IPC::Message& message) {
   base::AutoLock auto_lock(channel_lock_);
-  if (enabled_) {
-    if (message.type() == GpuCommandBufferMsg_WaitForTokenInRange::ID ||
-        message.type() == GpuCommandBufferMsg_WaitForGetOffsetInRange::ID) {
-      channel_->PostHandleOutOfOrderMessage(message);
-      return true;
-    }
+  DCHECK(channel_);
+  uint32_t order_num = sync_point_order_data_->GenerateUnprocessedOrderNumber();
+  std::unique_ptr<GpuChannelMessage> msg(
+      new GpuChannelMessage(message, order_num, base::TimeTicks::Now()));
 
-    uint32_t order_num =
-        sync_point_order_data_->GenerateUnprocessedOrderNumber();
-    std::unique_ptr<GpuChannelMessage> msg(
-        new GpuChannelMessage(message, order_num, base::TimeTicks::Now()));
+  channel_messages_.push_back(std::move(msg));
 
-    if (channel_messages_.empty()) {
-      DCHECK(scheduled_);
-      channel_->PostHandleMessage();
-    }
+  bool first_message = channel_messages_.size() == 1;
+  if (first_message)
+    PostHandleMessageOnQueue();
 
-    channel_messages_.push_back(std::move(msg));
+  if (preempting_flag_)
+    UpdatePreemptionStateHelper();
+}
 
-    if (preempting_flag_)
-      UpdatePreemptionStateHelper();
-
-    return true;
-  }
-  return false;
+void GpuChannelMessageQueue::PostHandleMessageOnQueue() {
+  channel_lock_.AssertAcquired();
+  DCHECK(channel_);
+  DCHECK(scheduled_);
+  DCHECK(!channel_messages_.empty());
+  DCHECK(!handle_message_post_task_pending_);
+  handle_message_post_task_pending_ = true;
+  main_task_runner_->PostTask(
+      FROM_HERE,
+      base::Bind(&GpuChannel::HandleMessageOnQueue, channel_->AsWeakPtr()));
 }
 
 const GpuChannelMessage* GpuChannelMessageQueue::BeginMessageProcessing() {
   base::AutoLock auto_lock(channel_lock_);
-  DCHECK(enabled_);
+  DCHECK(channel_);
+  DCHECK(scheduled_);
+  DCHECK(!channel_messages_.empty());
+  handle_message_post_task_pending_ = false;
   // If we have been preempted by another channel, just post a task to wake up.
   if (preempted_flag_ && preempted_flag_->IsSet()) {
-    channel_->PostHandleMessage();
+    PostHandleMessageOnQueue();
     return nullptr;
   }
-  if (channel_messages_.empty())
-    return nullptr;
   sync_point_order_data_->BeginProcessingOrderNumber(
       channel_messages_.front()->order_number);
   return channel_messages_.front().get();
 }
 
 void GpuChannelMessageQueue::PauseMessageProcessing() {
   base::AutoLock auto_lock(channel_lock_);
   DCHECK(!channel_messages_.empty());
 
   // If we have been preempted by another channel, just post a task to wake up.
   if (scheduled_)
-    channel_->PostHandleMessage();
+    PostHandleMessageOnQueue();
 
   sync_point_order_data_->PauseProcessingOrderNumber(
       channel_messages_.front()->order_number);
 }
 
 void GpuChannelMessageQueue::FinishMessageProcessing() {
   base::AutoLock auto_lock(channel_lock_);
   DCHECK(!channel_messages_.empty());
   DCHECK(scheduled_);
 
   sync_point_order_data_->FinishProcessingOrderNumber(
       channel_messages_.front()->order_number);
   channel_messages_.pop_front();
 
   if (!channel_messages_.empty())
-    channel_->PostHandleMessage();
+    PostHandleMessageOnQueue();
 
   if (preempting_flag_) {
     io_task_runner_->PostTask(
         FROM_HERE,
         base::Bind(&GpuChannelMessageQueue::UpdatePreemptionState, this));
   }
 }
 
 void GpuChannelMessageQueue::UpdatePreemptionState() {
   DCHECK(io_thread_checker_.CalledOnValidThread());
@@ skipping 180 matching lines @@
   channel_lock_.AssertAcquired();
   DCHECK(preemption_state_ == CHECKING || preemption_state_ == PREEMPTING);
   DCHECK(!scheduled_);
 
   preemption_state_ = WOULD_PREEMPT_DESCHEDULED;
   preempting_flag_->Reset();
   TRACE_COUNTER_ID1("gpu", "GpuChannel::Preempting", this, 0);
 }
 
 GpuChannelMessageFilter::GpuChannelMessageFilter(
-    scoped_refptr<GpuChannelMessageQueue> message_queue)
-    : message_queue_(std::move(message_queue)),
-      channel_(nullptr),
-      peer_pid_(base::kNullProcessId) {}
+    GpuChannel* gpu_channel,
+    scoped_refptr<GpuChannelMessageQueue> message_queue,
+    scoped_refptr<base::SingleThreadTaskRunner> main_task_runner)
+    : gpu_channel_(gpu_channel),
+      message_queue_(std::move(message_queue)),
+      main_task_runner_(std::move(main_task_runner)) {}
 
-GpuChannelMessageFilter::~GpuChannelMessageFilter() {}
+GpuChannelMessageFilter::~GpuChannelMessageFilter() {
+  DCHECK(!gpu_channel_);
+}
+
+void GpuChannelMessageFilter::Destroy() {
+  base::AutoLock auto_lock(gpu_channel_lock_);
+  gpu_channel_ = nullptr;
+}
 
 void GpuChannelMessageFilter::OnFilterAdded(IPC::Channel* channel) {
-  DCHECK(!channel_);
-  channel_ = channel;
-  for (scoped_refptr<IPC::MessageFilter>& filter : channel_filters_) {
-    filter->OnFilterAdded(channel_);
-  }
+  DCHECK(!ipc_channel_);
+  ipc_channel_ = channel;
+  for (scoped_refptr<IPC::MessageFilter>& filter : channel_filters_)
+    filter->OnFilterAdded(ipc_channel_);
 }
 
 void GpuChannelMessageFilter::OnFilterRemoved() {
-  for (scoped_refptr<IPC::MessageFilter>& filter : channel_filters_) {
+  for (scoped_refptr<IPC::MessageFilter>& filter : channel_filters_)
     filter->OnFilterRemoved();
-  }
-  channel_ = nullptr;
+  ipc_channel_ = nullptr;
   peer_pid_ = base::kNullProcessId;
 }
 
 void GpuChannelMessageFilter::OnChannelConnected(int32_t peer_pid) {
   DCHECK(peer_pid_ == base::kNullProcessId);
   peer_pid_ = peer_pid;
-  for (scoped_refptr<IPC::MessageFilter>& filter : channel_filters_) {
+  for (scoped_refptr<IPC::MessageFilter>& filter : channel_filters_)
     filter->OnChannelConnected(peer_pid);
-  }
 }
 
 void GpuChannelMessageFilter::OnChannelError() {
-  for (scoped_refptr<IPC::MessageFilter>& filter : channel_filters_) {
+  for (scoped_refptr<IPC::MessageFilter>& filter : channel_filters_)
     filter->OnChannelError();
-  }
 }
 
 void GpuChannelMessageFilter::OnChannelClosing() {
-  for (scoped_refptr<IPC::MessageFilter>& filter : channel_filters_) {
+  for (scoped_refptr<IPC::MessageFilter>& filter : channel_filters_)
     filter->OnChannelClosing();
-  }
 }
 
 void GpuChannelMessageFilter::AddChannelFilter(
     scoped_refptr<IPC::MessageFilter> filter) {
   channel_filters_.push_back(filter);
-  if (channel_)
-    filter->OnFilterAdded(channel_);
+  if (ipc_channel_)
+    filter->OnFilterAdded(ipc_channel_);
   if (peer_pid_ != base::kNullProcessId)
     filter->OnChannelConnected(peer_pid_);
 }
 
 void GpuChannelMessageFilter::RemoveChannelFilter(
     scoped_refptr<IPC::MessageFilter> filter) {
-  if (channel_)
+  if (ipc_channel_)
     filter->OnFilterRemoved();
-  channel_filters_.erase(
-      std::find(channel_filters_.begin(), channel_filters_.end(), filter));
+  base::Erase(channel_filters_, filter);
 }
 
 bool GpuChannelMessageFilter::OnMessageReceived(const IPC::Message& message) {
-  DCHECK(channel_);
+  DCHECK(ipc_channel_);
 
   if (message.should_unblock() || message.is_reply())
     return MessageErrorHandler(message, "Unexpected message type");
 
   if (message.type() == GpuChannelMsg_Nop::ID) {
     IPC::Message* reply = IPC::SyncMessage::GenerateReply(&message);
     Send(reply);
     return true;
   }
 
   for (scoped_refptr<IPC::MessageFilter>& filter : channel_filters_) {
     if (filter->OnMessageReceived(message))
       return true;
   }
 
-  if (!message_queue_->PushBackMessage(message))
+  base::AutoLock auto_lock(gpu_channel_lock_);
+  if (!gpu_channel_)
     return MessageErrorHandler(message, "Channel destroyed");
 
+  if (message.routing_id() == MSG_ROUTING_CONTROL ||
+      message.type() == GpuCommandBufferMsg_WaitForTokenInRange::ID ||
+      message.type() == GpuCommandBufferMsg_WaitForGetOffsetInRange::ID) {
+    // It's OK to post task that may never run even for sync messages, because
+    // if the channel is destroyed, the client Send will fail.
+    main_task_runner_->PostTask(FROM_HERE,
+                                base::Bind(&GpuChannel::HandleOutOfOrderMessage,
+                                           gpu_channel_->AsWeakPtr(), message));
+  } else {
+    // Message queue takes care of PostTask.
+    message_queue_->PushBackMessage(message);
+  }
+
   return true;
 }
 
 bool GpuChannelMessageFilter::Send(IPC::Message* message) {
-  return channel_->Send(message);
+  return ipc_channel_->Send(message);
 }
 
 bool GpuChannelMessageFilter::MessageErrorHandler(const IPC::Message& message,
                                                   const char* error_msg) {
   DLOG(ERROR) << error_msg;
   if (message.is_sync()) {
     IPC::Message* reply = IPC::SyncMessage::GenerateReply(&message);
     reply->set_reply_error();
     Send(reply);
   }
@@ skipping 14 matching lines @@
                        PreemptionFlag* preempting_flag,
                        PreemptionFlag* preempted_flag,
                        base::SingleThreadTaskRunner* task_runner,
                        base::SingleThreadTaskRunner* io_task_runner,
                        int32_t client_id,
                        uint64_t client_tracing_id,
                        bool allow_view_command_buffers,
                        bool allow_real_time_streams)
     : gpu_channel_manager_(gpu_channel_manager),
       sync_point_manager_(sync_point_manager),
-      unhandled_message_listener_(nullptr),
       preempting_flag_(preempting_flag),
       preempted_flag_(preempted_flag),
       client_id_(client_id),
       client_tracing_id_(client_tracing_id),
       task_runner_(task_runner),
       io_task_runner_(io_task_runner),
       share_group_(share_group),
       mailbox_manager_(mailbox),
       watchdog_(watchdog),
       allow_view_command_buffers_(allow_view_command_buffers),
       allow_real_time_streams_(allow_real_time_streams),
       weak_factory_(this) {
   DCHECK(gpu_channel_manager);
   DCHECK(client_id);
 
-  message_queue_ =
-      GpuChannelMessageQueue::Create(this, io_task_runner, preempting_flag,
-                                     preempted_flag, sync_point_manager);
+  message_queue_ = new GpuChannelMessageQueue(this, task_runner, io_task_runner,
+                                              preempting_flag, preempted_flag,
+                                              sync_point_manager);
 
-  filter_ = new GpuChannelMessageFilter(message_queue_);
+  filter_ = new GpuChannelMessageFilter(this, message_queue_, task_runner);
 }
 
 GpuChannel::~GpuChannel() {
   // Clear stubs first because of dependencies.
   stubs_.clear();
 
-  message_queue_->Disable();
+  // Destroy filter first so that no message queue gets no more messages.
+  filter_->Destroy();
+
+  message_queue_->Destroy();
 
   if (preempting_flag_.get())
     preempting_flag_->Reset();
 }
 
 IPC::ChannelHandle GpuChannel::Init(base::WaitableEvent* shutdown_event) {
   DCHECK(shutdown_event);
   DCHECK(!channel_);
 
   mojo::MessagePipe pipe;
@@ skipping 94 matching lines @@
                         OnCreateCommandBuffer)
     IPC_MESSAGE_HANDLER(GpuChannelMsg_DestroyCommandBuffer,
                         OnDestroyCommandBuffer)
     IPC_MESSAGE_HANDLER(GpuChannelMsg_GetDriverBugWorkArounds,
                         OnGetDriverBugWorkArounds)
     IPC_MESSAGE_UNHANDLED(handled = false)
   IPC_END_MESSAGE_MAP()
   return handled;
 }
 
-void GpuChannel::PostHandleMessage() {
-  task_runner_->PostTask(FROM_HERE, base::Bind(&GpuChannel::HandleMessage,
-                                               weak_factory_.GetWeakPtr()));
-}
-
-void GpuChannel::PostHandleOutOfOrderMessage(const IPC::Message& msg) {
-  task_runner_->PostTask(FROM_HERE,
-                         base::Bind(&GpuChannel::HandleOutOfOrderMessage,
-                                    weak_factory_.GetWeakPtr(), msg));
-}
-
-void GpuChannel::HandleMessage() {
+void GpuChannel::HandleMessageOnQueue() {
   const GpuChannelMessage* channel_msg =
       message_queue_->BeginMessageProcessing();
   if (!channel_msg)
     return;
 
   const IPC::Message& msg = channel_msg->message;
   int32_t routing_id = msg.routing_id();
   GpuCommandBufferStub* stub = LookupCommandBuffer(routing_id);
 
   DCHECK(!stub || stub->IsScheduled());
@@ skipping 172 matching lines @@
 #undef GPU_OP
 }
 
 void GpuChannel::CacheShader(const std::string& key,
                              const std::string& shader) {
   gpu_channel_manager_->delegate()->StoreShaderToDisk(client_id_, key, shader);
 }
 
 void GpuChannel::AddFilter(IPC::MessageFilter* filter) {
   io_task_runner_->PostTask(
-      FROM_HERE, base::Bind(&GpuChannelMessageFilter::AddChannelFilter, filter_,
-                            make_scoped_refptr(filter)));
+      FROM_HERE,
+      base::Bind(&GpuChannelMessageFilter::AddChannelFilter, filter_, filter));
 }
 
 void GpuChannel::RemoveFilter(IPC::MessageFilter* filter) {
   io_task_runner_->PostTask(
       FROM_HERE, base::Bind(&GpuChannelMessageFilter::RemoveChannelFilter,
-                            filter_, make_scoped_refptr(filter)));
+                            filter_, filter));
 }
 
 uint64_t GpuChannel::GetMemoryUsage() {
   // Collect the unique memory trackers in use by the |stubs_|.
   std::set<gles2::MemoryTracker*> unique_memory_trackers;
   for (auto& kv : stubs_)
     unique_memory_trackers.insert(kv.second->GetMemoryTracker());
 
   // Sum the memory usage for all unique memory trackers.
   uint64_t size = 0;
@@ skipping 31 matching lines @@
 
       return manager->gpu_memory_buffer_factory()
           ->AsImageFactory()
           ->CreateImageForGpuMemoryBuffer(handle, size, format, internalformat,
                                           client_id_, surface_handle);
     }
   }
 }
 
 }  // namespace gpu