WIP GPU scheduler + delayed activation / tile draw

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 21 matching lines @@
 #include "build/build_config.h"
 #include "gpu/command_buffer/common/mailbox.h"
 #include "gpu/command_buffer/service/command_executor.h"
 #include "gpu/command_buffer/service/image_factory.h"
 #include "gpu/command_buffer/service/mailbox_manager.h"
 #include "gpu/command_buffer/service/sync_point_manager.h"
 #include "gpu/ipc/common/gpu_messages.h"
 #include "gpu/ipc/service/gpu_channel_manager.h"
 #include "gpu/ipc/service/gpu_channel_manager_delegate.h"
 #include "gpu/ipc/service/gpu_memory_buffer_factory.h"
+#include "gpu/ipc/service/gpu_scheduler.h"
 #include "ipc/ipc_channel.h"
 #include "ipc/message_filter.h"
 #include "ui/gl/gl_context.h"
 #include "ui/gl/gl_image_shared_memory.h"
 #include "ui/gl/gl_surface.h"
 
 namespace gpu {
 namespace {
 
 // Number of milliseconds between successive vsync. Many GL commands block
@@ skipping 12 matching lines @@
 // Stop the preemption once the time for the longest pending IPC drops
 // below this threshold.
 const int64_t kStopPreemptThresholdMs = kVsyncIntervalMs;
 
 }  // anonymous namespace
 
 scoped_refptr<GpuChannelMessageQueue> GpuChannelMessageQueue::Create(
     int32_t stream_id,
     GpuStreamPriority stream_priority,
     GpuChannel* channel,
-    const scoped_refptr<base::SingleThreadTaskRunner>& io_task_runner,
-    const scoped_refptr<PreemptionFlag>& preempting_flag,
-    const scoped_refptr<PreemptionFlag>& preempted_flag,
+    scoped_refptr<base::SingleThreadTaskRunner> main_task_runner,
+    scoped_refptr<base::SingleThreadTaskRunner> io_task_runner,
+    GpuScheduler* scheduler,
+    scoped_refptr<PreemptionFlag> preempting_flag,
+    scoped_refptr<PreemptionFlag> preemption_flag,
     SyncPointManager* sync_point_manager) {
-  return new GpuChannelMessageQueue(stream_id, stream_priority, channel,
-                                    io_task_runner, preempting_flag,
-                                    preempted_flag, sync_point_manager);
+  return new GpuChannelMessageQueue(
+      stream_id, stream_priority, channel, std::move(main_task_runner),
+      std::move(io_task_runner), scheduler, std::move(preempting_flag),
+      std::move(preemption_flag), sync_point_manager);
 }
 
 scoped_refptr<SyncPointOrderData>
 GpuChannelMessageQueue::GetSyncPointOrderData() {
   return sync_point_order_data_;
 }
 
 GpuChannelMessageQueue::GpuChannelMessageQueue(
     int32_t stream_id,
     GpuStreamPriority stream_priority,
     GpuChannel* channel,
-    const scoped_refptr<base::SingleThreadTaskRunner>& io_task_runner,
-    const scoped_refptr<PreemptionFlag>& preempting_flag,
-    const scoped_refptr<PreemptionFlag>& preempted_flag,
+    scoped_refptr<base::SingleThreadTaskRunner> main_task_runner,
+    scoped_refptr<base::SingleThreadTaskRunner> io_task_runner,
+    GpuScheduler* scheduler,
+    scoped_refptr<PreemptionFlag> preempting_flag,
+    scoped_refptr<PreemptionFlag> preemption_flag,
     SyncPointManager* sync_point_manager)
     : stream_id_(stream_id),
-      stream_priority_(stream_priority),
-      enabled_(true),
+      priority_(stream_priority),
+      num_routes_(0u),
       scheduled_(true),
       channel_(channel),
       preemption_state_(IDLE),
       max_preemption_time_(
           base::TimeDelta::FromMilliseconds(kMaxPreemptTimeMs)),
       timer_(new base::OneShotTimer),
       sync_point_order_data_(SyncPointOrderData::Create()),
-      io_task_runner_(io_task_runner),
-      preempting_flag_(preempting_flag),
-      preempted_flag_(preempted_flag),
+      main_task_runner_(std::move(main_task_runner)),
+      io_task_runner_(std::move(io_task_runner)),
+      scheduler_(scheduler),
+      preempting_flag_(std::move(preempting_flag)),
+      preemption_flag_(std::move(preemption_flag)),
       sync_point_manager_(sync_point_manager) {
   timer_->SetTaskRunner(io_task_runner);
   io_thread_checker_.DetachFromThread();
+  if (scheduler_)
+    scheduler_->AddStream(this, stream_priority);
 }
 
 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 queues will no longer be modified when |Destroy| is
+  // called therefore it is now safe to modify the queue without the lock.
   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();
   }
 
   sync_point_order_data_->Destroy();
   sync_point_order_data_ = nullptr;
 
+  if (scheduler_)
+    scheduler_->RemoveStream(this);
+
+  channel_ = nullptr;
+
   io_task_runner_->PostTask(
-      FROM_HERE, base::Bind(&GpuChannelMessageQueue::DisableIO, this));
+      FROM_HERE, base::Bind(&GpuChannelMessageQueue::DestroyIO, this));
 }
 
-void GpuChannelMessageQueue::DisableIO() {
+void GpuChannelMessageQueue::DestroyIO() {
   DCHECK(io_thread_checker_.CalledOnValidThread());
   timer_ = nullptr;
 }
 
+void GpuChannelMessageQueue::OnRouteAdded() {
+  num_routes_++;
+}
+
+void GpuChannelMessageQueue::OnRouteRemoved() {
+  num_routes_--;
+}
+
+size_t GpuChannelMessageQueue::NumRoutes() const {
+  return num_routes_;
+}
+
+void GpuChannelMessageQueue::Run() {
+  DCHECK(scheduler_);
+  while (!scheduler_->ShouldYield() && HasMessages() && IsScheduled())
+    channel_->HandleMessageOnStream(this);
+}
+
 bool GpuChannelMessageQueue::IsScheduled() const {
   base::AutoLock lock(channel_lock_);
   return scheduled_;
 }
 
-void GpuChannelMessageQueue::OnRescheduled(bool scheduled) {
+void GpuChannelMessageQueue::Schedule() {
   base::AutoLock lock(channel_lock_);
-  DCHECK(enabled_);
-  if (scheduled_ == scheduled)
+  if (scheduled_)
     return;
-  scheduled_ = scheduled;
-  if (scheduled)
-    channel_->PostHandleMessage(this);
-  if (preempting_flag_) {
-    io_task_runner_->PostTask(
-        FROM_HERE,
-        base::Bind(&GpuChannelMessageQueue::UpdatePreemptionState, this));
+  scheduled_ = true;
+  if (scheduler_) {
+    scheduler_->ScheduleStream(this);
+  } else {
+    if (!channel_messages_.empty())
+      channel_->PostHandleMessageOnStream(this);
+    if (preempting_flag_)
+      PostUpdatePreemptionState();
   }
 }
 
+void GpuChannelMessageQueue::Deschedule() {
+  base::AutoLock lock(channel_lock_);
+  if (!scheduled_)
+    return;
+  scheduled_ = false;
+  if (scheduler_) {
+    scheduler_->DescheduleStream(this);
+  } else if (preempting_flag_) {
+    PostUpdatePreemptionState();
+  }
+}
+
 uint32_t GpuChannelMessageQueue::GetUnprocessedOrderNum() const {
   return sync_point_order_data_->unprocessed_order_num();
 }
 
 uint32_t GpuChannelMessageQueue::GetProcessedOrderNum() const {
   return sync_point_order_data_->processed_order_num();
 }
 
-bool GpuChannelMessageQueue::PushBackMessage(const IPC::Message& message) {
+bool GpuChannelMessageQueue::HasMessages() const {
   base::AutoLock auto_lock(channel_lock_);
-  if (enabled_) {
-    if (message.type() == GpuCommandBufferMsg_WaitForTokenInRange::ID ||
-        message.type() == GpuCommandBufferMsg_WaitForGetOffsetInRange::ID) {
-      channel_->PostHandleOutOfOrderMessage(message);
-      return true;
-    }
+  return !channel_messages_.empty();
+}
 
-    uint32_t order_num = sync_point_order_data_->GenerateUnprocessedOrderNumber(
-        sync_point_manager_);
-    std::unique_ptr<GpuChannelMessage> msg(
-        new GpuChannelMessage(message, order_num, base::TimeTicks::Now()));
+void GpuChannelMessageQueue::PushBackMessage(const IPC::Message& message) {
+  base::AutoLock auto_lock(channel_lock_);
+  DCHECK(channel_);
 
-    if (channel_messages_.empty()) {
-      DCHECK(scheduled_);
-      channel_->PostHandleMessage(this);
-    }
+  uint32_t order_num = sync_point_order_data_->GenerateUnprocessedOrderNumber(
+      sync_point_manager_);
+  std::unique_ptr<GpuChannelMessage> msg(
+      new GpuChannelMessage(message, order_num, base::TimeTicks::Now()));
+  channel_messages_.push_back(std::move(msg));
 
-    channel_messages_.push_back(std::move(msg));
+  bool start_running = channel_messages_.size() == 1 && scheduled_;
 
+  if (scheduler_) {
+    if (start_running)
+      scheduler_->ScheduleStream(this);
+  } else {
+    if (start_running)
+      channel_->PostHandleMessageOnStream(this);
     if (preempting_flag_)
       UpdatePreemptionStateHelper();
-
-    return true;
   }
-  return false;
 }
 
 const GpuChannelMessage* GpuChannelMessageQueue::BeginMessageProcessing() {
   base::AutoLock auto_lock(channel_lock_);
-  DCHECK(enabled_);
-  // If we have been preempted by another channel, just post a task to wake up.
-  if (preempted_flag_ && preempted_flag_->IsSet()) {
-    channel_->PostHandleMessage(this);
+  if (channel_messages_.empty())
+    return nullptr;
+  if (preemption_flag_ && preemption_flag_->IsSet()) {
+    channel_->PostHandleMessageOnStream(this);
     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(this);
-
   sync_point_order_data_->PauseProcessingOrderNumber(
       channel_messages_.front()->order_number);
+
+  if (!scheduler_) {
+    if (scheduled_)
+      channel_->PostHandleMessageOnStream(this);
+    if (preempting_flag_)
+      PostUpdatePreemptionState();
+  }
 }
 
 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(this);
-
-  if (preempting_flag_) {
-    io_task_runner_->PostTask(
-        FROM_HERE,
-        base::Bind(&GpuChannelMessageQueue::UpdatePreemptionState, this));
+  if (scheduler_) {
+    if (channel_messages_.empty())
+      scheduler_->DescheduleStream(this);
+  } else {
+    if (!channel_messages_.empty())
+      channel_->PostHandleMessageOnStream(this);
+    if (preempting_flag_)
+      PostUpdatePreemptionState();
   }
 }
 
+void GpuChannelMessageQueue::PostUpdatePreemptionState() {
+  DCHECK(preempting_flag_);
+  DCHECK(!scheduler_);
+  io_task_runner_->PostTask(
+      FROM_HERE,
+      base::Bind(&GpuChannelMessageQueue::UpdatePreemptionState, this));
+}
+
 void GpuChannelMessageQueue::UpdatePreemptionState() {
-  DCHECK(io_thread_checker_.CalledOnValidThread());
-  DCHECK(preempting_flag_);
   base::AutoLock lock(channel_lock_);
   UpdatePreemptionStateHelper();
 }
 
 void GpuChannelMessageQueue::UpdatePreemptionStateHelper() {
   DCHECK(io_thread_checker_.CalledOnValidThread());
+  channel_lock_.AssertAcquired();
   DCHECK(preempting_flag_);
-  channel_lock_.AssertAcquired();
   switch (preemption_state_) {
     case IDLE:
       UpdateStateIdle();
       break;
     case WAITING:
       UpdateStateWaiting();
       break;
     case CHECKING:
       UpdateStateChecking();
       break;
@@ skipping 160 matching lines @@
   DCHECK(preempting_flag_);
   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()
-    : channel_(nullptr), peer_pid_(base::kNullProcessId) {}
+GpuChannelMessageFilter::GpuChannelMessageFilter(GpuChannel* gpu_channel)
+    : gpu_channel_(gpu_channel) {}
 
 GpuChannelMessageFilter::~GpuChannelMessageFilter() {}
 
+scoped_refptr<GpuChannelMessageFilter> GpuChannelMessageFilter::Create(
+    GpuChannel* gpu_channel) {
+  return make_scoped_refptr(new GpuChannelMessageFilter(gpu_channel));
+}
+
+void GpuChannelMessageFilter::Disable() {
+  base::AutoLock lock(lock_);
+  enabled_ = false;
+}
+
 void GpuChannelMessageFilter::OnFilterAdded(IPC::Channel* channel) {
   DCHECK(!channel_);
   channel_ = channel;
   for (scoped_refptr<IPC::MessageFilter>& filter : channel_filters_) {
     filter->OnFilterAdded(channel_);
   }
 }
 
 void GpuChannelMessageFilter::OnFilterRemoved() {
   for (scoped_refptr<IPC::MessageFilter>& filter : channel_filters_) {
@@ skipping 33 matching lines @@
 }
 
 void GpuChannelMessageFilter::RemoveChannelFilter(
     scoped_refptr<IPC::MessageFilter> filter) {
   if (channel_)
     filter->OnFilterRemoved();
   channel_filters_.erase(
       std::find(channel_filters_.begin(), channel_filters_.end(), filter));
 }
 
-// This gets called from the main thread and assumes that all messages which
-// lead to creation of a new route are synchronous messages.
-// TODO(sunnyps): Create routes (and streams) on the IO thread so that we can
-// make the CreateCommandBuffer/VideoDecoder/VideoEncoder messages asynchronous.
 void GpuChannelMessageFilter::AddRoute(
     int32_t route_id,
-    const scoped_refptr<GpuChannelMessageQueue>& queue) {
-  base::AutoLock lock(routes_lock_);
-  routes_.insert(std::make_pair(route_id, queue));
+    scoped_refptr<GpuChannelMessageQueue> stream) {
+  base::AutoLock lock(lock_);
+  routes_.insert(std::make_pair(route_id, stream));
 }
 
-void GpuChannelMessageFilter::RemoveRoute(int32_t route_id) {
-  base::AutoLock lock(routes_lock_);
-  routes_.erase(route_id);
+scoped_refptr<GpuChannelMessageQueue> GpuChannelMessageFilter::RemoveRoute(
+    int32_t route_id) {
+  base::AutoLock lock(lock_);
+  auto it = routes_.find(route_id);
+  if (it != routes_.end()) {
+    scoped_refptr<GpuChannelMessageQueue> stream(it->second);
+    routes_.erase(it);
+    return stream;
+  }
+  return nullptr;
 }
 
 bool GpuChannelMessageFilter::OnMessageReceived(const IPC::Message& message) {
+  base::AutoLock lock(lock_);
   DCHECK(channel_);
 
+  if (!enabled_)
+    return MessageErrorHandler(message, "Channel destroyed");
+
   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;
   }
 
-  scoped_refptr<GpuChannelMessageQueue> message_queue =
-      LookupStreamByRoute(message.routing_id());
+  if (message.routing_id() == MSG_ROUTING_CONTROL ||
+      message.type() == GpuCommandBufferMsg_WaitForTokenInRange::ID ||
+      message.type() == GpuCommandBufferMsg_WaitForGetOffsetInRange::ID) {
+    gpu_channel_->PostHandleOutOfOrderMessage(message);
+    return true;
+  }
 
-  if (!message_queue)
-    return MessageErrorHandler(message, "Could not find message queue");
+  scoped_refptr<GpuChannelMessageQueue> stream =
+      LookupRoute(message.routing_id());
+  if (!stream)
+    return MessageErrorHandler(message, "Could not find route");
 
-  if (!message_queue->PushBackMessage(message))
-    return MessageErrorHandler(message, "Channel destroyed");
-
+  stream->PushBackMessage(message);
   return true;
 }
 
 bool GpuChannelMessageFilter::Send(IPC::Message* message) {
   return channel_->Send(message);
 }
 
-scoped_refptr<GpuChannelMessageQueue>
-GpuChannelMessageFilter::LookupStreamByRoute(int32_t route_id) {
-  base::AutoLock lock(routes_lock_);
+scoped_refptr<GpuChannelMessageQueue> GpuChannelMessageFilter::LookupRoute(
+    int32_t route_id) {
+  scoped_refptr<GpuChannelMessageQueue> stream;
   auto it = routes_.find(route_id);
   if (it != routes_.end())
-    return it->second;
-  return nullptr;
+    stream = it->second;
+  return stream;
 }
 
 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);
   }
   return true;
 }
 
-GpuChannel::GpuChannel(GpuChannelManager* gpu_channel_manager,
-                       SyncPointManager* sync_point_manager,
-                       GpuWatchdogThread* watchdog,
-                       gl::GLShareGroup* share_group,
-                       gles2::MailboxManager* mailbox,
-                       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)
+GpuChannel::GpuChannel(
+    GpuChannelManager* gpu_channel_manager,
+    SyncPointManager* sync_point_manager,
+    GpuWatchdogThread* watchdog,
+    GpuScheduler* scheduler,
+    scoped_refptr<gl::GLShareGroup> share_group,
+    scoped_refptr<gles2::MailboxManager> mailbox_manager,
+    scoped_refptr<PreemptionFlag> preempting_flag,
+    scoped_refptr<PreemptionFlag> preemption_flag,
+    scoped_refptr<base::SingleThreadTaskRunner> task_runner,
+    scoped_refptr<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),
+      scheduler_(scheduler),
       unhandled_message_listener_(nullptr),
-      preempting_flag_(preempting_flag),
-      preempted_flag_(preempted_flag),
+      preempting_flag_(std::move(preempting_flag)),
+      preemption_flag_(std::move(preemption_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),
+      task_runner_(std::move(task_runner)),
+      io_task_runner_(std::move(io_task_runner)),
+      share_group_(std::move(share_group)),
+      mailbox_manager_(std::move(mailbox_manager)),
       watchdog_(watchdog),
       allow_view_command_buffers_(allow_view_command_buffers),
-      allow_real_time_streams_(allow_real_time_streams),
-      weak_factory_(this) {
+      allow_real_time_streams_(allow_real_time_streams) {
   DCHECK(gpu_channel_manager);
   DCHECK(client_id);
-
-  filter_ = new GpuChannelMessageFilter();
-
-  scoped_refptr<GpuChannelMessageQueue> control_queue =
-      CreateStream(GPU_STREAM_DEFAULT, GpuStreamPriority::HIGH);
-  AddRouteToStream(MSG_ROUTING_CONTROL, GPU_STREAM_DEFAULT);
+  filter_ = GpuChannelMessageFilter::Create(this);
 }
 
 GpuChannel::~GpuChannel() {
   // Clear stubs first because of dependencies.
   stubs_.clear();
 
+  filter_->Disable();
+
   for (auto& kv : streams_)
-    kv.second->Disable();
+    kv.second->Destroy();
 
-  if (preempting_flag_.get())
+  if (preempting_flag_)
     preempting_flag_->Reset();
 }
 
 IPC::ChannelHandle GpuChannel::Init(base::WaitableEvent* shutdown_event) {
   DCHECK(shutdown_event);
   DCHECK(!channel_);
 
   mojo::MessagePipe pipe;
   channel_ = IPC::SyncChannel::Create(pipe.handle0.release(),
                                       IPC::Channel::MODE_SERVER, this,
                                       io_task_runner_, false, shutdown_event);
 
   channel_->AddFilter(filter_.get());
 
   return pipe.handle1.release();
 }
 
 void GpuChannel::SetUnhandledMessageListener(IPC::Listener* listener) {
   unhandled_message_listener_ = listener;
 }
 
-base::WeakPtr<GpuChannel> GpuChannel::AsWeakPtr() {
-  return weak_factory_.GetWeakPtr();
-}
-
 base::ProcessId GpuChannel::GetClientPID() const {
   DCHECK_NE(peer_pid_, base::kNullProcessId);
   return peer_pid_;
 }
 
+base::Callback<bool(void)> GpuChannel::GetPreemptionCallback() const {
+  if (scheduler_) {
+    return base::Bind(&GpuScheduler::ShouldYield, base::Unretained(scheduler_));
+  } else if (preemption_flag_) {
+    return base::Bind(&PreemptionFlag::IsSet,
+                      base::RetainedRef(preemption_flag_));
+  }
+  return base::Callback<bool(void)>();
+}
+
 uint32_t GpuChannel::GetProcessedOrderNum() const {
   uint32_t processed_order_num = 0;
   for (auto& kv : streams_) {
     processed_order_num =
         std::max(processed_order_num, kv.second->GetProcessedOrderNum());
   }
   return processed_order_num;
 }
 
 uint32_t GpuChannel::GetUnprocessedOrderNum() const {
   uint32_t unprocessed_order_num = 0;
   for (auto& kv : streams_) {
     unprocessed_order_num =
         std::max(unprocessed_order_num, kv.second->GetUnprocessedOrderNum());
   }
   return unprocessed_order_num;
 }
 
 bool GpuChannel::OnMessageReceived(const IPC::Message& msg) {
-  // All messages should be pushed to channel_messages_ and handled separately.
+  // All messages should be pushed to channel_messages_ and handled
+  // separately.
   NOTREACHED();
   return false;
 }
 
 void GpuChannel::OnChannelConnected(int32_t peer_pid) {
   peer_pid_ = peer_pid;
 }
 
 void GpuChannel::OnChannelError() {
   gpu_channel_manager_->RemoveChannel(client_id_);
 }
 
 bool GpuChannel::Send(IPC::Message* message) {
   // The GPU process must never send a synchronous IPC message to the renderer
   // process. This could result in deadlock.
   DCHECK(!message->is_sync());
 
   DVLOG(1) << "sending message @" << message << " on channel @" << this
            << " with type " << message->type();
 
   if (!channel_) {
     delete message;
     return false;
   }
 
   return channel_->Send(message);
 }
 
-void GpuChannel::OnStreamRescheduled(int32_t stream_id, bool scheduled) {
-  scoped_refptr<GpuChannelMessageQueue> queue = LookupStream(stream_id);
-  DCHECK(queue);
-  queue->OnRescheduled(scheduled);
+void GpuChannel::ScheduleCommandBuffer(GpuCommandBufferStub* stub) {
+  scoped_refptr<GpuChannelMessageQueue> stream =
+      LookupStream(stub->stream_id());
+  DCHECK(stream);
+  stream->Schedule();
+}
+
+void GpuChannel::DescheduleCommandBuffer(GpuCommandBufferStub* stub) {
+  scoped_refptr<GpuChannelMessageQueue> stream =
+      LookupStream(stub->stream_id());
+  DCHECK(stream);
+  stream->Deschedule();
+}
+
+scoped_refptr<GpuChannelMessageQueue> GpuChannel::LookupStreamByCommandBufferId(
+    CommandBufferId command_buffer_id) {
+  int32_t channel_id = GpuCommandBufferStub::GetChannelID(command_buffer_id);
+  GpuChannel* channel = gpu_channel_manager_->LookupChannel(channel_id);
+
+  int32_t route_id = GpuCommandBufferStub::GetRouteID(command_buffer_id);
+  GpuCommandBufferStub* stub = channel->LookupCommandBuffer(route_id);
+
+  return channel->LookupStream(stub->stream_id());
 }
 
 GpuCommandBufferStub* GpuChannel::LookupCommandBuffer(int32_t route_id) {
-  return stubs_.get(route_id);
+  auto it = stubs_.find(route_id);
+  if (it != stubs_.end())
+    return it->second.get();
+  return nullptr;
 }
 
 void GpuChannel::LoseAllContexts() {
   gpu_channel_manager_->LoseAllContexts();
 }
 
 void GpuChannel::MarkAllContextsLost() {
   for (auto& kv : stubs_)
     kv.second->MarkContextLost();
 }
 
 bool GpuChannel::AddRoute(int32_t route_id,
                           int32_t stream_id,
                           IPC::Listener* listener) {
   if (router_.AddRoute(route_id, listener)) {
-    AddRouteToStream(route_id, stream_id);
+    scoped_refptr<GpuChannelMessageQueue> stream = LookupStream(stream_id);
+    filter_->AddRoute(route_id, stream);
+    stream->OnRouteAdded();
     return true;
   }
   return false;
 }
 
 void GpuChannel::RemoveRoute(int32_t route_id) {
   router_.RemoveRoute(route_id);
-  RemoveRouteFromStream(route_id);
+  scoped_refptr<GpuChannelMessageQueue> stream = filter_->RemoveRoute(route_id);
+  stream->OnRouteRemoved();
+  if (stream->NumRoutes() == 0) {
+    stream->Destroy();
+    streams_.erase(stream->stream_id());
+  }
 }
 
 bool GpuChannel::OnControlMessageReceived(const IPC::Message& msg) {
   bool handled = true;
   IPC_BEGIN_MESSAGE_MAP(GpuChannel, msg)
     IPC_MESSAGE_HANDLER(GpuChannelMsg_CreateCommandBuffer,
                         OnCreateCommandBuffer)
     IPC_MESSAGE_HANDLER(GpuChannelMsg_DestroyCommandBuffer,
                         OnDestroyCommandBuffer)
     IPC_MESSAGE_HANDLER(GpuChannelMsg_GetDriverBugWorkArounds,
                         OnGetDriverBugWorkArounds)
     IPC_MESSAGE_UNHANDLED(handled = false)
   IPC_END_MESSAGE_MAP()
   return handled;
 }
 
 scoped_refptr<SyncPointOrderData> GpuChannel::GetSyncPointOrderData(
     int32_t stream_id) {
   auto it = streams_.find(stream_id);
   DCHECK(it != streams_.end());
   DCHECK(it->second);
   return it->second->GetSyncPointOrderData();
 }
 
-void GpuChannel::PostHandleMessage(
-    const scoped_refptr<GpuChannelMessageQueue>& queue) {
-  task_runner_->PostTask(FROM_HERE,
-                         base::Bind(&GpuChannel::HandleMessage,
-                                    weak_factory_.GetWeakPtr(), queue));
+void GpuChannel::PostHandleMessageOnStream(
+    scoped_refptr<GpuChannelMessageQueue> stream) {
+  task_runner_->PostTask(
+      FROM_HERE,
+      base::Bind(&GpuChannel::HandleMessageOnStream, AsWeakPtr(), stream));
 }
 
-void GpuChannel::PostHandleOutOfOrderMessage(const IPC::Message& msg) {
-  task_runner_->PostTask(FROM_HERE,
-                         base::Bind(&GpuChannel::HandleOutOfOrderMessage,
-                                    weak_factory_.GetWeakPtr(), msg));
-}
-
-void GpuChannel::HandleMessage(
-    const scoped_refptr<GpuChannelMessageQueue>& message_queue) {
-  const GpuChannelMessage* channel_msg =
-      message_queue->BeginMessageProcessing();
-  if (!channel_msg)
+void GpuChannel::HandleMessageOnStream(
+    scoped_refptr<GpuChannelMessageQueue> stream) {
+  const GpuChannelMessage* msg_wrapper = stream->BeginMessageProcessing();
+  if (!msg_wrapper)
     return;
 
-  const IPC::Message& msg = channel_msg->message;
-  int32_t routing_id = msg.routing_id();
-  GpuCommandBufferStub* stub = stubs_.get(routing_id);
-
+  const IPC::Message& msg = msg_wrapper->message;
+  GpuCommandBufferStub* stub = LookupCommandBuffer(msg.routing_id());
   DCHECK(!stub || stub->IsScheduled());
-
   DVLOG(1) << "received message @" << &msg << " on channel @" << this
            << " with type " << msg.type();
 
   HandleMessageHelper(msg);
 
-  // If we get descheduled or yield while processing a message.
-  if ((stub && stub->HasUnprocessedCommands()) ||
-      !message_queue->IsScheduled()) {
+  // If we get descheduled or preempted while processing a message.
+  if ((stub && stub->HasUnprocessedCommands()) || !stream->IsScheduled()) {
     DCHECK((uint32_t)GpuCommandBufferMsg_AsyncFlush::ID == msg.type() ||
            (uint32_t)GpuCommandBufferMsg_WaitSyncToken::ID == msg.type());
-    message_queue->PauseMessageProcessing();
+    stream->PauseMessageProcessing();
   } else {
-    message_queue->FinishMessageProcessing();
+    stream->FinishMessageProcessing();
   }
 }
 
+void GpuChannel::PostHandleOutOfOrderMessage(const IPC::Message& msg) {
+  task_runner_->PostTask(
+      FROM_HERE,
+      base::Bind(&GpuChannel::HandleOutOfOrderMessage, AsWeakPtr(), msg));
+}
+
+void GpuChannel::HandleOutOfOrderMessage(const IPC::Message& msg) {
+  HandleMessageHelper(msg);
+}
+
 void GpuChannel::HandleMessageHelper(const IPC::Message& msg) {
   int32_t routing_id = msg.routing_id();
-
   bool handled = false;
   if (routing_id == MSG_ROUTING_CONTROL) {
     handled = OnControlMessageReceived(msg);
   } else {
     handled = router_.RouteMessage(msg);
   }
 
   if (!handled && unhandled_message_listener_)
     handled = unhandled_message_listener_->OnMessageReceived(msg);
 
   // Respond to sync messages even if router failed to route.
   if (!handled && msg.is_sync()) {
     IPC::Message* reply = IPC::SyncMessage::GenerateReply(&msg);
     reply->set_reply_error();
     Send(reply);
   }
 }
 
-void GpuChannel::HandleOutOfOrderMessage(const IPC::Message& msg) {
-  HandleMessageHelper(msg);
-}
-
-void GpuChannel::HandleMessageForTesting(const IPC::Message& msg) {
-  HandleMessageHelper(msg);
-}
-
 scoped_refptr<GpuChannelMessageQueue> GpuChannel::CreateStream(
     int32_t stream_id,
     GpuStreamPriority stream_priority) {
-  DCHECK(streams_.find(stream_id) == streams_.end());
-  scoped_refptr<GpuChannelMessageQueue> queue = GpuChannelMessageQueue::Create(
-      stream_id, stream_priority, this, io_task_runner_,
-      (stream_id == GPU_STREAM_DEFAULT) ? preempting_flag_ : nullptr,
-      preempted_flag_, sync_point_manager_);
-  streams_.insert(std::make_pair(stream_id, queue));
-  streams_to_num_routes_.insert(std::make_pair(stream_id, 0));
-  return queue;
+  return GpuChannelMessageQueue::Create(
+      stream_id, stream_priority, this, task_runner_, io_task_runner_,
+      scheduler_, preempting_flag_, preemption_flag_, sync_point_manager_);
 }
 
 scoped_refptr<GpuChannelMessageQueue> GpuChannel::LookupStream(
     int32_t stream_id) {
-  auto stream_it = streams_.find(stream_id);
-  if (stream_it != streams_.end())
-    return stream_it->second;
+  auto it = streams_.find(stream_id);
+  if (it != streams_.end())
+    return it->second;
   return nullptr;
 }
 
-void GpuChannel::DestroyStreamIfNecessary(
-    const scoped_refptr<GpuChannelMessageQueue>& queue) {
-  int32_t stream_id = queue->stream_id();
-  if (streams_to_num_routes_[stream_id] == 0) {
-    queue->Disable();
-    streams_to_num_routes_.erase(stream_id);
-    streams_.erase(stream_id);
-  }
-}
-
-void GpuChannel::AddRouteToStream(int32_t route_id, int32_t stream_id) {
-  DCHECK(streams_.find(stream_id) != streams_.end());
-  DCHECK(routes_to_streams_.find(route_id) == routes_to_streams_.end());
-  streams_to_num_routes_[stream_id]++;
-  routes_to_streams_.insert(std::make_pair(route_id, stream_id));
-  filter_->AddRoute(route_id, streams_[stream_id]);
-}
-
-void GpuChannel::RemoveRouteFromStream(int32_t route_id) {
-  DCHECK(routes_to_streams_.find(route_id) != routes_to_streams_.end());
-  int32_t stream_id = routes_to_streams_[route_id];
-  DCHECK(streams_.find(stream_id) != streams_.end());
-  routes_to_streams_.erase(route_id);
-  streams_to_num_routes_[stream_id]--;
-  filter_->RemoveRoute(route_id);
-  DestroyStreamIfNecessary(streams_[stream_id]);
-}
-
 #if defined(OS_ANDROID)
 const GpuCommandBufferStub* GpuChannel::GetOneStub() const {
   for (const auto& kv : stubs_) {
-    const GpuCommandBufferStub* stub = kv.second;
+    const GpuCommandBufferStub* stub = kv.second.get();
     if (stub->decoder() && !stub->decoder()->WasContextLost())
       return stub;
   }
   return nullptr;
 }
 #endif
 
 void GpuChannel::OnCreateCommandBuffer(
     const GPUCreateCommandBufferConfig& init_params,
     int32_t route_id,
     base::SharedMemoryHandle shared_state_handle,
     bool* result,
     gpu::Capabilities* capabilities) {
   TRACE_EVENT2("gpu", "GpuChannel::OnCreateCommandBuffer", "route_id", route_id,
                "offscreen", (init_params.surface_handle == kNullSurfaceHandle));
   std::unique_ptr<base::SharedMemory> shared_state_shm(
       new base::SharedMemory(shared_state_handle, false));
-  std::unique_ptr<GpuCommandBufferStub> stub =
+  GpuCommandBufferStub* stub =
       CreateCommandBuffer(init_params, route_id, std::move(shared_state_shm));
   if (stub) {
     *result = true;
     *capabilities = stub->decoder()->GetCapabilities();
-    stubs_.set(route_id, std::move(stub));
   } else {
     *result = false;
     *capabilities = gpu::Capabilities();
   }
 }
 
-std::unique_ptr<GpuCommandBufferStub> GpuChannel::CreateCommandBuffer(
+GpuCommandBufferStub* GpuChannel::CreateCommandBuffer(
     const GPUCreateCommandBufferConfig& init_params,
     int32_t route_id,
     std::unique_ptr<base::SharedMemory> shared_state_shm) {
   if (init_params.surface_handle != kNullSurfaceHandle &&
       !allow_view_command_buffers_) {
     DLOG(ERROR) << "GpuChannel::CreateCommandBuffer(): attempt to create a "
                    "view context on a non-priviledged channel";
     return nullptr;
   }
 
   int32_t share_group_id = init_params.share_group_id;
-  GpuCommandBufferStub* share_group = stubs_.get(share_group_id);
+  GpuCommandBufferStub* share_group = LookupCommandBuffer(share_group_id);
 
   if (!share_group && share_group_id != MSG_ROUTING_NONE) {
-    DLOG(ERROR)
-        << "GpuChannel::CreateCommandBuffer(): invalid share group id";
+    DLOG(ERROR) << "GpuChannel::CreateCommandBuffer(): invalid share group id";
     return nullptr;
   }
 
   int32_t stream_id = init_params.stream_id;
   if (share_group && stream_id != share_group->stream_id()) {
     DLOG(ERROR) << "GpuChannel::CreateCommandBuffer(): stream id does not "
                    "match share group stream id";
     return nullptr;
   }
 
@@ skipping 12 matching lines @@
                    "not initialized";
     return nullptr;
   }
 
   if (share_group && share_group->decoder()->WasContextLost()) {
     DLOG(ERROR) << "GpuChannel::CreateCommandBuffer(): shared context was "
                    "already lost";
     return nullptr;
   }
 
-  scoped_refptr<GpuChannelMessageQueue> queue = LookupStream(stream_id);
-  if (!queue)
-    queue = CreateStream(stream_id, stream_priority);
+  scoped_refptr<GpuChannelMessageQueue> stream = LookupStream(stream_id);
+  if (!stream) {
+    stream = CreateStream(stream_id, stream_priority);
+    streams_.insert(std::make_pair(stream_id, stream));
+  }
 
   std::unique_ptr<GpuCommandBufferStub> stub(GpuCommandBufferStub::Create(
       this, share_group, init_params, route_id, std::move(shared_state_shm)));
 
-  if (!stub) {
-    DestroyStreamIfNecessary(queue);
+  if (!router_.AddRoute(route_id, stub.get())) {
+    DLOG(ERROR) << "GpuChannel::CreateCommandBuffer(): failed to add route";
+    if (stream->NumRoutes() == 0) {
+      stream->Destroy();
+      streams_.erase(stream_id);
+    }
     return nullptr;
   }
 
-  if (!AddRoute(route_id, stream_id, stub.get())) {
-    DestroyStreamIfNecessary(queue);
-    DLOG(ERROR) << "GpuChannel::CreateCommandBuffer(): failed to add route";
-    return nullptr;
-  }
+  filter_->AddRoute(route_id, stream);
+  stream->OnRouteAdded();
 
-  return stub;
+  GpuCommandBufferStub* ret_stub = stub.get();
+  stubs_.insert(std::make_pair(route_id, std::move(stub)));
+
+  return ret_stub;
 }
 
 void GpuChannel::OnDestroyCommandBuffer(int32_t route_id) {
-  TRACE_EVENT1("gpu", "GpuChannel::OnDestroyCommandBuffer",
-               "route_id", route_id);
-
-  std::unique_ptr<GpuCommandBufferStub> stub = stubs_.take_and_erase(route_id);
-  // In case the renderer is currently blocked waiting for a sync reply from the
-  // stub, we need to make sure to reschedule the correct stream here.
-  if (stub && !stub->IsScheduled()) {
-    // This stub won't get a chance to reschedule the stream so do that now.
-    OnStreamRescheduled(stub->stream_id(), true);
+  TRACE_EVENT1("gpu", "GpuChannel::OnDestroyCommandBuffer", "route_id",
+               route_id);
+  auto it = stubs_.find(route_id);
+  if (it != stubs_.end()) {
+    std::unique_ptr<GpuCommandBufferStub> stub(std::move(it->second));
+    stubs_.erase(it);
+    // In case the renderer is currently blocked waiting for a sync reply from
+    // the stub, we need to make sure to reschedule the correct stream here.
+    ScheduleCommandBuffer(stub.get());
   }
-
   RemoveRoute(route_id);
 }
 
 void GpuChannel::OnGetDriverBugWorkArounds(
     std::vector<std::string>* gpu_driver_bug_workarounds) {
   gpu_driver_bug_workarounds->clear();
 #define GPU_OP(type, name)                                     \
   if (gpu_channel_manager_->gpu_driver_bug_workarounds().name) \
     gpu_driver_bug_workarounds->push_back(#name);
   GPU_DRIVER_BUG_WORKAROUNDS(GPU_OP)
 #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,
+                 base::RetainedRef(filter_), make_scoped_refptr(filter)));
 }
 
 void GpuChannel::RemoveFilter(IPC::MessageFilter* filter) {
   io_task_runner_->PostTask(
-      FROM_HERE, base::Bind(&GpuChannelMessageFilter::RemoveChannelFilter,
-                            filter_, make_scoped_refptr(filter)));
+      FROM_HERE,
+      base::Bind(&GpuChannelMessageFilter::RemoveChannelFilter,
+                 base::RetainedRef(filter_), make_scoped_refptr(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