WIP GPU scheduler + delayed activation / tile draw

gpu/ipc/service/gpu_channel.cc

Index: gpu/ipc/service/gpu_channel.cc
diff --git a/gpu/ipc/service/gpu_channel.cc b/gpu/ipc/service/gpu_channel.cc
index 8ed788a38174e3cf06d81a61d7c6181fe5a81986..871e34401678fd9141e4a10de019d0611ad222de 100644
--- a/gpu/ipc/service/gpu_channel.cc
+++ b/gpu/ipc/service/gpu_channel.cc
@@ -39,6 +39,7 @@
 #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"
@@ -71,13 +72,16 @@ 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>
@@ -89,13 +93,15 @@ 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),
@@ -103,30 +109,25 @@ GpuChannelMessageQueue::GpuChannelMessageQueue(
           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()) {
@@ -140,32 +141,67 @@ void GpuChannelMessageQueue::Disable() {
   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();
   }
 }
 
@@ -177,45 +213,42 @@ 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);
-    return nullptr;
-  }
   if (channel_messages_.empty())
     return nullptr;
+  if (preemption_flag_ && preemption_flag_->IsSet()) {
+    channel_->PostHandleMessageOnStream(this);
+    return nullptr;
+  }
   sync_point_order_data_->BeginProcessingOrderNumber(
       channel_messages_.front()->order_number);
   return channel_messages_.front().get();
@@ -225,12 +258,15 @@ 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() {
@@ -242,27 +278,34 @@ void GpuChannelMessageQueue::FinishMessageProcessing() {
       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::UpdatePreemptionState() {
-  DCHECK(io_thread_checker_.CalledOnValidThread());
+void GpuChannelMessageQueue::PostUpdatePreemptionState() {
   DCHECK(preempting_flag_);
+  DCHECK(!scheduler_);
+  io_task_runner_->PostTask(
+      FROM_HERE,
+      base::Bind(&GpuChannelMessageQueue::UpdatePreemptionState, this));
+}
+
+void GpuChannelMessageQueue::UpdatePreemptionState() {
   base::AutoLock lock(channel_lock_);
   UpdatePreemptionStateHelper();
 }
 
 void GpuChannelMessageQueue::UpdatePreemptionStateHelper() {
   DCHECK(io_thread_checker_.CalledOnValidThread());
-  DCHECK(preempting_flag_);
   channel_lock_.AssertAcquired();
+  DCHECK(preempting_flag_);
   switch (preemption_state_) {
     case IDLE:
       UpdateStateIdle();
@@ -443,11 +486,21 @@ void GpuChannelMessageQueue::TransitionToWouldPreemptDescheduled() {
   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;
@@ -501,25 +554,32 @@ void GpuChannelMessageFilter::RemoveChannelFilter(
       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");
 
@@ -534,15 +594,19 @@ bool GpuChannelMessageFilter::OnMessageReceived(const IPC::Message& message) {
       return true;
   }
 
-  scoped_refptr<GpuChannelMessageQueue> message_queue =
-      LookupStreamByRoute(message.routing_id());
-
-  if (!message_queue)
-    return MessageErrorHandler(message, "Could not find message queue");
+  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->PushBackMessage(message))
-    return MessageErrorHandler(message, "Channel destroyed");
+  scoped_refptr<GpuChannelMessageQueue> stream =
+      LookupRoute(message.routing_id());
+  if (!stream)
+    return MessageErrorHandler(message, "Could not find route");
 
+  stream->PushBackMessage(message);
   return true;
 }
 
@@ -550,13 +614,13 @@ 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,
@@ -570,52 +634,51 @@ bool GpuChannelMessageFilter::MessageErrorHandler(const IPC::Message& message,
   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();
 }
 
@@ -637,15 +700,21 @@ 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_) {
@@ -665,7 +734,8 @@ uint32_t GpuChannel::GetUnprocessedOrderNum() const {
 }
 
 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;
 }
@@ -694,14 +764,36 @@ bool GpuChannel::Send(IPC::Message* message) {
   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() {
@@ -717,7 +809,9 @@ 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;
@@ -725,7 +819,12 @@ bool GpuChannel::AddRoute(int32_t route_id,
 
 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) {
@@ -750,51 +849,49 @@ scoped_refptr<SyncPointOrderData> GpuChannel::GetSyncPointOrderData(
   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);
@@ -813,67 +910,26 @@ void GpuChannel::HandleMessageHelper(const IPC::Message& msg) {
   }
 }
 
-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;
   }
@@ -891,19 +947,18 @@ void GpuChannel::OnCreateCommandBuffer(
                "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) {
@@ -915,11 +970,10 @@ std::unique_ptr<GpuCommandBufferStub> GpuChannel::CreateCommandBuffer(
   }
 
   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;
   }
 
@@ -952,39 +1006,44 @@ std::unique_ptr<GpuCommandBufferStub> GpuChannel::CreateCommandBuffer(
     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);
-    return nullptr;
-  }
-
-  if (!AddRoute(route_id, stream_id, stub.get())) {
-    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;
   }
 
-  return stub;
+  filter_->AddRoute(route_id, stream);
+  stream->OnRouteAdded();
+
+  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);
 }
 
@@ -1005,14 +1064,16 @@ void GpuChannel::CacheShader(const std::string& key,
 
 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() {