GPU Channel's now maintain a global order number for each processed IPC.

content/common/gpu/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.
 
 #if defined(OS_WIN)
 #include <windows.h>
 #endif
 
 #include "content/common/gpu/gpu_channel.h"
 
 #include <algorithm>
-#include <queue>
+#include <deque>
+#include <set>
 #include <vector>
 
+#include "base/atomicops.h"
 #include "base/bind.h"
 #include "base/command_line.h"
 #include "base/location.h"
 #include "base/single_thread_task_runner.h"
 #include "base/stl_util.h"
 #include "base/strings/string_util.h"
+#include "base/synchronization/lock.h"
 #include "base/thread_task_runner_handle.h"
 #include "base/timer/timer.h"
 #include "base/trace_event/memory_dump_manager.h"
 #include "base/trace_event/process_memory_dump.h"
 #include "base/trace_event/trace_event.h"
 #include "content/common/gpu/gpu_channel_manager.h"
 #include "content/common/gpu/gpu_memory_buffer_factory.h"
 #include "content/common/gpu/gpu_messages.h"
 #include "content/common/gpu/media/gpu_jpeg_decode_accelerator.h"
 #include "content/public/common/content_switches.h"
@@ skipping 27 matching lines @@
 const int64 kPreemptWaitTimeMs = 2 * kVsyncIntervalMs;
 
 // Once we trigger a preemption, the maximum duration that we will wait
 // before clearing the preemption.
 const int64 kMaxPreemptTimeMs = kVsyncIntervalMs;
 
 // Stop the preemption once the time for the longest pending IPC drops
 // below this threshold.
 const int64 kStopPreemptThresholdMs = kVsyncIntervalMs;
 
+const uint32_t kOutOfOrderNumber = -1u;
+
 }  // anonymous namespace
 
+struct GpuChannelMessage {
+  uint32_t order_number;
+  base::TimeTicks time_received;
+  IPC::Message message;
+
+  // TODO(dyen): Temporary sync point data, remove once new sync point lands.
+  bool retire_sync_point;
+  uint32 sync_point_number;
+
+  GpuChannelMessage(uint32_t order_num, const IPC::Message& msg)
+      : order_number(order_num),
+        time_received(base::TimeTicks::Now()),
+        message(msg),
+        retire_sync_point(false),
+        sync_point_number(0) {}
+};
+
+class GpuChannelMessageQueue
+    : public base::RefCountedThreadSafe<GpuChannelMessageQueue> {
+ public:
+  static scoped_refptr<GpuChannelMessageQueue> Create(
+      base::WeakPtr<GpuChannel> gpu_channel,
+      scoped_refptr<base::SingleThreadTaskRunner> task_runner) {
+    return new GpuChannelMessageQueue(gpu_channel, task_runner);
+  }
+
+  uint32_t GetUnprocessedOrderNum() {
+    base::AutoLock auto_lock(channel_messages_lock_);
+    return unprocessed_order_num_;
+  }
+
+  void PushBackMessage(uint32_t order_number, const IPC::Message& message) {
+    base::AutoLock auto_lock(channel_messages_lock_);
+    if (enabled_) {
+      PushMessageHelper(order_number,
+                        new GpuChannelMessage(order_number, message));
+    }
+  }
+
+  void PushOutOfOrderMessage(const IPC::Message& message) {
+    // These are pushed out of order so should not have any order messages.
+    base::AutoLock auto_lock(channel_messages_lock_);
+    if (enabled_) {
+      PushOutOfOrderHelper(new GpuChannelMessage(kOutOfOrderNumber, message));
+    }
+  }
+
+  bool GenerateSyncPointMessage(gpu::SyncPointManager* sync_point_manager,
+                                uint32_t order_number,
+                                const IPC::Message& message,
+                                bool retire_sync_point,
+                                uint32_t* sync_point_number) {
+    DCHECK(message.type() == GpuCommandBufferMsg_InsertSyncPoint::ID);
+    base::AutoLock auto_lock(channel_messages_lock_);
+    if (enabled_) {
+      const uint32 sync_point = sync_point_manager->GenerateSyncPoint();
+
+      GpuChannelMessage* msg = new GpuChannelMessage(order_number, message);
+      msg->retire_sync_point = retire_sync_point;
+      msg->sync_point_number = sync_point;
+
+      *sync_point_number = sync_point;
+      PushMessageHelper(order_number, msg);
+      return true;
+    }
+    return false;
+  }
+
+  bool HasQueuedMessages() {
+    base::AutoLock auto_lock(channel_messages_lock_);
+    return !channel_messages_.empty();
+  }
+
+  base::TimeTicks GetNextMessageTimeTick() {
+    base::AutoLock auto_lock(channel_messages_lock_);
+
+    base::TimeTicks next_message_tick;
+    if (!channel_messages_.empty())
+      next_message_tick = channel_messages_.front()->time_received;
+
+    base::TimeTicks next_out_of_order_tick;
+    if (!out_of_order_messages_.empty())
+      next_out_of_order_tick = out_of_order_messages_.front()->time_received;
+
+    if (next_message_tick.is_null())
+      return next_out_of_order_tick;
+    else if (next_out_of_order_tick.is_null())
+      return next_message_tick;
+    else
+      return std::min(next_message_tick, next_out_of_order_tick);
+  }
+
+ protected:
+  virtual ~GpuChannelMessageQueue() {}
+
+ private:
+  friend class GpuChannel;
+  friend class base::RefCountedThreadSafe<GpuChannelMessageQueue>;
+
+  GpuChannelMessageQueue(
+      base::WeakPtr<GpuChannel> gpu_channel,
+      scoped_refptr<base::SingleThreadTaskRunner> task_runner)
+      : enabled_(true),
+        unprocessed_order_num_(0),
+        gpu_channel_(gpu_channel),
+        task_runner_(task_runner) {}
+
+  void DeleteAndDisableMessages(GpuChannelManager* gpu_channel_manager) {
+    {
+      base::AutoLock auto_lock(channel_messages_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.
+    while (!channel_messages_.empty()) {
+      GpuChannelMessage* msg = channel_messages_.front();
+      // This needs to clean up both GpuCommandBufferMsg_InsertSyncPoint and
+      // GpuCommandBufferMsg_RetireSyncPoint messages, safer to just check
+      // if we have a sync point number here.
+      if (msg->sync_point_number) {
+        gpu_channel_manager->sync_point_manager()->RetireSyncPoint(
+            msg->sync_point_number);
+      }
+      delete msg;
+      channel_messages_.pop_front();
+    }
+    STLDeleteElements(&out_of_order_messages_);
+  }
+
+  void PushUnfinishedMessage(uint32_t order_number,
+                             const IPC::Message& message) {
+    // This is pushed only if it was unfinished, so order number is kept.
+    GpuChannelMessage* msg = new GpuChannelMessage(order_number, message);
+    base::AutoLock auto_lock(channel_messages_lock_);
+    DCHECK(enabled_);
+    const bool was_empty =
+        (channel_messages_.empty() && out_of_order_messages_.empty());
+    if (order_number == kOutOfOrderNumber)
+      out_of_order_messages_.push_front(msg);
+    else
+      channel_messages_.push_front(msg);
+
+    if (was_empty)
+      ScheduleHandleMessage();
+  }
+
+  void ScheduleHandleMessage() {
+    task_runner_->PostTask(
+        FROM_HERE, base::Bind(&GpuChannel::HandleMessage, gpu_channel_));
+  }
+
+  void PushMessageHelper(uint32_t order_number, GpuChannelMessage* msg) {
+    channel_messages_lock_.AssertAcquired();
+    DCHECK(enabled_);
+    unprocessed_order_num_ = order_number;
+    const bool was_empty =
+        channel_messages_.empty() && out_of_order_messages_.empty();
+    channel_messages_.push_back(msg);
+    if (was_empty)
+      ScheduleHandleMessage();
+  }
+
+  void PushOutOfOrderHelper(GpuChannelMessage* msg) {
+    channel_messages_lock_.AssertAcquired();
+    DCHECK(enabled_);
+    const bool was_empty =
+        channel_messages_.empty() && out_of_order_messages_.empty();
+    out_of_order_messages_.push_back(msg);
+    if (was_empty)
+      ScheduleHandleMessage();
+  }
+
+  bool enabled_;
+
+  // Highest IPC order number seen, set when queued on the IO thread.
+  uint32_t unprocessed_order_num_;
+  std::deque<GpuChannelMessage*> channel_messages_;
+  std::deque<GpuChannelMessage*> out_of_order_messages_;
+
+  // This lock protects enabled_, unprocessed_order_num_, and both deques.
+  base::Lock channel_messages_lock_;
+
+  base::WeakPtr<GpuChannel> gpu_channel_;
+  scoped_refptr<base::SingleThreadTaskRunner> task_runner_;
+
+  DISALLOW_COPY_AND_ASSIGN(GpuChannelMessageQueue);
+};
+
 // This filter does three things:
 // - it counts and timestamps each message forwarded to the channel
 //   so that we can preempt other channels if a message takes too long to
 //   process. To guarantee fairness, we must wait a minimum amount of time
 //   before preempting and we limit the amount of time that we can preempt in
 //   one shot (see constants above).
 // - it handles the GpuCommandBufferMsg_InsertSyncPoint message on the IO
 //   thread, generating the sync point ID and responding immediately, and then
 //   posting a task to insert the GpuCommandBufferMsg_RetireSyncPoint message
 //   into the channel's queue.
 // - it generates mailbox names for clients of the GPU process on the IO thread.
 class GpuChannelMessageFilter : public IPC::MessageFilter {
  public:
   GpuChannelMessageFilter(
-      base::WeakPtr<GpuChannel> gpu_channel,
+      scoped_refptr<GpuChannelMessageQueue> message_queue,
       gpu::SyncPointManager* sync_point_manager,
       scoped_refptr<base::SingleThreadTaskRunner> task_runner,
       bool future_sync_points)
       : preemption_state_(IDLE),
-        gpu_channel_(gpu_channel),
+        message_queue_(message_queue),
         sender_(nullptr),
         sync_point_manager_(sync_point_manager),
         task_runner_(task_runner),
-        messages_forwarded_to_channel_(0),
         a_stub_is_descheduled_(false),
         future_sync_points_(future_sync_points) {}
 
   void OnFilterAdded(IPC::Sender* sender) override {
     DCHECK(!sender_);
     sender_ = sender;
     timer_ = make_scoped_ptr(new base::OneShotTimer<GpuChannelMessageFilter>);
   }
 
   void OnFilterRemoved() override {
     DCHECK(sender_);
     sender_ = nullptr;
     timer_ = nullptr;
   }
 
   bool OnMessageReceived(const IPC::Message& message) override {
     DCHECK(sender_);
 
+    const uint32_t order_number = global_order_counter_++;
     bool handled = false;
     if ((message.type() == GpuCommandBufferMsg_RetireSyncPoint::ID) &&
         !future_sync_points_) {
       DLOG(ERROR) << "Untrusted client should not send "
                      "GpuCommandBufferMsg_RetireSyncPoint message";
       return true;
     }
 
     if (message.type() == GpuCommandBufferMsg_InsertSyncPoint::ID) {
       base::Tuple<bool> retire;
       IPC::Message* reply = IPC::SyncMessage::GenerateReply(&message);
       if (!GpuCommandBufferMsg_InsertSyncPoint::ReadSendParam(&message,
                                                               &retire)) {
         reply->set_reply_error();
         Send(reply);
         return true;
       }
       if (!future_sync_points_ && !base::get<0>(retire)) {
         LOG(ERROR) << "Untrusted contexts can't create future sync points";
         reply->set_reply_error();
         Send(reply);
         return true;
       }
-      uint32 sync_point = sync_point_manager_->GenerateSyncPoint();
+
+      // Message queue must handle the entire sync point generation because the
+      // message queue could be disabled from the main thread during generation.
+      uint32_t sync_point = 0u;
+      if (!message_queue_->GenerateSyncPointMessage(
+              sync_point_manager_, order_number, message, base::get<0>(retire),
+              &sync_point)) {
+        LOG(ERROR) << "GpuChannel has been destroyed.";
+        reply->set_reply_error();
+        Send(reply);
+        return true;
+      }
+
+      DCHECK_NE(sync_point, 0u);
       GpuCommandBufferMsg_InsertSyncPoint::WriteReplyParams(reply, sync_point);
       Send(reply);
-      task_runner_->PostTask(
-          FROM_HERE,
-          base::Bind(&GpuChannelMessageFilter::InsertSyncPointOnMainThread,
-                     gpu_channel_, sync_point_manager_, message.routing_id(),
-                     base::get<0>(retire), sync_point));
       handled = true;
     }
 
     // These are handled by GpuJpegDecodeAccelerator and
     // GpuVideoDecodeAccelerator.
     // TODO(kcwu) Modify GpuChannel::AddFilter to handle additional filters by
     // GpuChannelMessageFilter instead of by IPC::SyncChannel directly. Then we
     // don't need to exclude them one by one here.
     if (message.type() == AcceleratedJpegDecoderMsg_Decode::ID ||
         message.type() == AcceleratedJpegDecoderMsg_Destroy::ID ||
         message.type() == AcceleratedVideoDecoderMsg_Decode::ID) {
       return false;
     }
 
-    // All other messages get processed by the GpuChannel.
-    messages_forwarded_to_channel_++;
-    if (preempting_flag_.get())
-      pending_messages_.push(PendingMessage(messages_forwarded_to_channel_));
+    // Forward all other messages to the GPU Channel.
+    if (!handled && !message.is_reply() && !message.should_unblock()) {
+      if (message.type() == GpuCommandBufferMsg_WaitForTokenInRange::ID ||
+          message.type() == GpuCommandBufferMsg_WaitForGetOffsetInRange::ID) {
+        // Move Wait commands to the head of the queue, so the renderer
+        // doesn't have to wait any longer than necessary.
+        message_queue_->PushOutOfOrderMessage(message);
+      } else {
+        message_queue_->PushBackMessage(order_number, message);
+      }
+      handled = true;
+    }
+
     UpdatePreemptionState();
-
     return handled;
   }
 
-  void MessageProcessed(uint64 messages_processed) {
-    while (!pending_messages_.empty() &&
-           pending_messages_.front().message_number <= messages_processed)
-      pending_messages_.pop();
-    UpdatePreemptionState();
-  }
+  void OnMessageProcessed() { UpdatePreemptionState(); }
 
   void SetPreemptingFlagAndSchedulingState(
       gpu::PreemptionFlag* preempting_flag,
       bool a_stub_is_descheduled) {
     preempting_flag_ = preempting_flag;
     a_stub_is_descheduled_ = a_stub_is_descheduled;
   }
 
   void UpdateStubSchedulingState(bool a_stub_is_descheduled) {
     a_stub_is_descheduled_ = a_stub_is_descheduled;
@@ skipping 23 matching lines @@
     // We would like to preempt, but some stub is descheduled.
     WOULD_PREEMPT_DESCHEDULED,
   };
 
   PreemptionState preemption_state_;
 
   // Maximum amount of time that we can spend in PREEMPTING.
   // It is reset when we transition to IDLE.
   base::TimeDelta max_preemption_time_;
 
-  struct PendingMessage {
-    uint64 message_number;
-    base::TimeTicks time_received;
-
-    explicit PendingMessage(uint64 message_number)
-        : message_number(message_number),
-          time_received(base::TimeTicks::Now()) {
-    }
-  };
-
   void UpdatePreemptionState() {
     switch (preemption_state_) {
       case IDLE:
-        if (preempting_flag_.get() && !pending_messages_.empty())
+        if (preempting_flag_.get() && message_queue_->HasQueuedMessages())
           TransitionToWaiting();
         break;
       case WAITING:
         // A timer will transition us to CHECKING.
         DCHECK(timer_->IsRunning());
         break;
-      case CHECKING:
-        if (!pending_messages_.empty()) {
-          base::TimeDelta time_elapsed =
-              base::TimeTicks::Now() - pending_messages_.front().time_received;
+      case CHECKING: {
+        base::TimeTicks time_tick = message_queue_->GetNextMessageTimeTick();
+        if (!time_tick.is_null()) {
+          base::TimeDelta time_elapsed = base::TimeTicks::Now() - time_tick;
           if (time_elapsed.InMilliseconds() < kPreemptWaitTimeMs) {
             // Schedule another check for when the IPC may go long.
             timer_->Start(
                 FROM_HERE,
                 base::TimeDelta::FromMilliseconds(kPreemptWaitTimeMs) -
                     time_elapsed,
                 this, &GpuChannelMessageFilter::UpdatePreemptionState);
           } else {
             if (a_stub_is_descheduled_)
               TransitionToWouldPreemptDescheduled();
             else
               TransitionToPreempting();
           }
         }
-        break;
+      } break;
       case PREEMPTING:
         // A TransitionToIdle() timer should always be running in this state.
         DCHECK(timer_->IsRunning());
         if (a_stub_is_descheduled_)
           TransitionToWouldPreemptDescheduled();
         else
           TransitionToIdleIfCaughtUp();
         break;
       case WOULD_PREEMPT_DESCHEDULED:
         // A TransitionToIdle() timer should never be running in this state.
         DCHECK(!timer_->IsRunning());
         if (!a_stub_is_descheduled_)
           TransitionToPreempting();
         else
           TransitionToIdleIfCaughtUp();
         break;
       default:
         NOTREACHED();
     }
   }
 
   void TransitionToIdleIfCaughtUp() {
     DCHECK(preemption_state_ == PREEMPTING ||
            preemption_state_ == WOULD_PREEMPT_DESCHEDULED);
-    if (pending_messages_.empty()) {
+    base::TimeTicks next_tick = message_queue_->GetNextMessageTimeTick();
+    if (next_tick.is_null()) {
       TransitionToIdle();
     } else {
-      base::TimeDelta time_elapsed =
-          base::TimeTicks::Now() - pending_messages_.front().time_received;
+      base::TimeDelta time_elapsed = base::TimeTicks::Now() - next_tick;
       if (time_elapsed.InMilliseconds() < kStopPreemptThresholdMs)
         TransitionToIdle();
     }
   }
 
   void TransitionToIdle() {
     DCHECK(preemption_state_ == PREEMPTING ||
            preemption_state_ == WOULD_PREEMPT_DESCHEDULED);
     // Stop any outstanding timer set to force us from PREEMPTING to IDLE.
     timer_->Stop();
@@ skipping 65 matching lines @@
       }
     }
 
     preemption_state_ = WOULD_PREEMPT_DESCHEDULED;
     preempting_flag_->Reset();
     TRACE_COUNTER_ID1("gpu", "GpuChannel::Preempting", this, 0);
 
     UpdatePreemptionState();
   }
 
-  static void InsertSyncPointOnMainThread(
-      base::WeakPtr<GpuChannel> gpu_channel,
-      gpu::SyncPointManager* manager,
-      int32 routing_id,
-      bool retire,
-      uint32 sync_point) {
-    // This function must ensure that the sync point will be retired. Normally
-    // we'll find the stub based on the routing ID, and associate the sync point
-    // with it, but if that fails for any reason (channel or stub already
-    // deleted, invalid routing id), we need to retire the sync point
-    // immediately.
-    if (gpu_channel) {
-      GpuCommandBufferStub* stub = gpu_channel->LookupCommandBuffer(routing_id);
-      if (stub) {
-        stub->AddSyncPoint(sync_point);
-        if (retire) {
-          GpuCommandBufferMsg_RetireSyncPoint message(routing_id, sync_point);
-          gpu_channel->OnMessageReceived(message);
-        }
-        return;
-      } else {
-        gpu_channel->MessageProcessed();
-      }
-    }
-    manager->RetireSyncPoint(sync_point);
-  }
-
-  // NOTE: this weak pointer is never dereferenced on the IO thread, it's only
-  // passed through - therefore the WeakPtr assumptions are respected.
-  base::WeakPtr<GpuChannel> gpu_channel_;
+  // The message_queue_ is used to handle messages on the main thread.
+  scoped_refptr<GpuChannelMessageQueue> message_queue_;
   IPC::Sender* sender_;
   gpu::SyncPointManager* sync_point_manager_;
   scoped_refptr<base::SingleThreadTaskRunner> task_runner_;
   scoped_refptr<gpu::PreemptionFlag> preempting_flag_;
 
-  std::queue<PendingMessage> pending_messages_;
-
-  // Count of the number of IPCs forwarded to the GpuChannel.
-  uint64 messages_forwarded_to_channel_;
-
   // This timer is created and destroyed on the IO thread.
   scoped_ptr<base::OneShotTimer<GpuChannelMessageFilter>> timer_;
 
   bool a_stub_is_descheduled_;
 
   // True if this channel can create future sync points.
   bool future_sync_points_;
+
+  // This number is only ever incremented/read on the IO thread.
+  static uint32_t global_order_counter_;
 };
 
+uint32_t GpuChannelMessageFilter::global_order_counter_ = 0;
+
 GpuChannel::GpuChannel(GpuChannelManager* gpu_channel_manager,
                        GpuWatchdog* watchdog,
                        gfx::GLShareGroup* share_group,
                        gpu::gles2::MailboxManager* mailbox,
                        base::SingleThreadTaskRunner* task_runner,
                        base::SingleThreadTaskRunner* io_task_runner,
                        int client_id,
                        uint64_t client_tracing_id,
                        bool software,
                        bool allow_future_sync_points)
     : gpu_channel_manager_(gpu_channel_manager),
       channel_id_(IPC::Channel::GenerateVerifiedChannelID("gpu")),
-      messages_processed_(0),
       client_id_(client_id),
       client_tracing_id_(client_tracing_id),
       task_runner_(task_runner),
       io_task_runner_(io_task_runner),
       share_group_(share_group ? share_group : new gfx::GLShareGroup),
       mailbox_manager_(mailbox
                            ? scoped_refptr<gpu::gles2::MailboxManager>(mailbox)
                            : gpu::gles2::MailboxManager::Create()),
       subscription_ref_set_(new gpu::gles2::SubscriptionRefSet),
       pending_valuebuffer_state_(new gpu::ValueStateMap),
       watchdog_(watchdog),
       software_(software),
-      handle_messages_scheduled_(false),
-      currently_processing_message_(nullptr),
+      current_order_num_(0),
+      processed_order_num_(0),
       num_stubs_descheduled_(0),
       allow_future_sync_points_(allow_future_sync_points),
       weak_factory_(this) {
   DCHECK(gpu_channel_manager);
   DCHECK(client_id);
 
+  message_queue_ =
+      GpuChannelMessageQueue::Create(weak_factory_.GetWeakPtr(), task_runner);
+
   filter_ = new GpuChannelMessageFilter(
-      weak_factory_.GetWeakPtr(), gpu_channel_manager_->sync_point_manager(),
-      task_runner_, allow_future_sync_points_);
+      message_queue_, gpu_channel_manager_->sync_point_manager(), task_runner_,
+      allow_future_sync_points_);
 
   subscription_ref_set_->AddObserver(this);
 }
 
 GpuChannel::~GpuChannel() {
   // Clear stubs first because of dependencies.
   stubs_.clear();
 
-  STLDeleteElements(&deferred_messages_);
+  message_queue_->DeleteAndDisableMessages(gpu_channel_manager_);
+
   subscription_ref_set_->RemoveObserver(this);
   if (preempting_flag_.get())
     preempting_flag_->Reset();
 }
 
 IPC::ChannelHandle GpuChannel::Init(base::WaitableEvent* shutdown_event,
                                     IPC::AttachmentBroker* attachment_broker) {
   DCHECK(shutdown_event);
   DCHECK(!channel_);
 
@@ skipping 14 matching lines @@
   channel_->AddFilter(filter_.get());
 
   return channel_handle;
 }
 
 base::ProcessId GpuChannel::GetClientPID() const {
   return channel_->GetPeerPID();
 }
 
 bool GpuChannel::OnMessageReceived(const IPC::Message& message) {
-  DVLOG(1) << "received message @" << &message << " on channel @" << this
-           << " with type " << message.type();
-
-  if (message.type() == GpuCommandBufferMsg_WaitForTokenInRange::ID ||
-      message.type() == GpuCommandBufferMsg_WaitForGetOffsetInRange::ID) {
-    // Move Wait commands to the head of the queue, so the renderer
-    // doesn't have to wait any longer than necessary.
-    deferred_messages_.push_front(new IPC::Message(message));
-  } else {
-    deferred_messages_.push_back(new IPC::Message(message));
-  }
-
-  OnScheduled();
-
-  return true;
+  // All messages should be pushed to channel_messages_ and handled separately.
+  NOTREACHED();
+  return false;
 }
 
 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());
@@ skipping 12 matching lines @@
 void GpuChannel::OnAddSubscription(unsigned int target) {
   gpu_channel_manager()->Send(
       new GpuHostMsg_AddSubscription(client_id_, target));
 }
 
 void GpuChannel::OnRemoveSubscription(unsigned int target) {
   gpu_channel_manager()->Send(
       new GpuHostMsg_RemoveSubscription(client_id_, target));
 }
 
-void GpuChannel::RequeueMessage() {
-  DCHECK(currently_processing_message_);
-  deferred_messages_.push_front(
-      new IPC::Message(*currently_processing_message_));
-  messages_processed_--;
-  currently_processing_message_ = NULL;
-}
-
-void GpuChannel::OnScheduled() {
-  if (handle_messages_scheduled_)
-    return;
-  // Post a task to handle any deferred messages. The deferred message queue is
-  // not emptied here, which ensures that OnMessageReceived will continue to
-  // defer newly received messages until the ones in the queue have all been
-  // handled by HandleMessage. HandleMessage is invoked as a
-  // task to prevent reentrancy.
-  task_runner_->PostTask(FROM_HERE, base::Bind(&GpuChannel::HandleMessage,
-                                               weak_factory_.GetWeakPtr()));
-  handle_messages_scheduled_ = true;
-}
-
 void GpuChannel::StubSchedulingChanged(bool scheduled) {
   bool a_stub_was_descheduled = num_stubs_descheduled_ > 0;
   if (scheduled) {
     num_stubs_descheduled_--;
-    OnScheduled();
+    message_queue_->ScheduleHandleMessage();
   } else {
     num_stubs_descheduled_++;
   }
   DCHECK_LE(num_stubs_descheduled_, stubs_.size());
   bool a_stub_is_descheduled = num_stubs_descheduled_ > 0;
 
   if (a_stub_is_descheduled != a_stub_was_descheduled) {
     if (preempting_flag_.get()) {
       io_task_runner_->PostTask(
           FROM_HERE,
@@ skipping 105 matching lines @@
                         OnDestroyCommandBuffer)
     IPC_MESSAGE_HANDLER_DELAY_REPLY(GpuMsg_CreateJpegDecoder,
                                     OnCreateJpegDecoder)
     IPC_MESSAGE_UNHANDLED(handled = false)
   IPC_END_MESSAGE_MAP()
   DCHECK(handled) << msg.type();
   return handled;
 }
 
 void GpuChannel::HandleMessage() {
-  handle_messages_scheduled_ = false;
-  if (deferred_messages_.empty())
+  GpuChannelMessage* m = nullptr;
+  GpuCommandBufferStub* stub = nullptr;
+  {
+    base::AutoLock auto_lock(message_queue_->channel_messages_lock_);
+    if (!message_queue_->out_of_order_messages_.empty()) {
+      m = message_queue_->out_of_order_messages_.front();
+      DCHECK(m->order_number == kOutOfOrderNumber);
+      message_queue_->out_of_order_messages_.pop_front();
+    } else if (!message_queue_->channel_messages_.empty()) {
+      m = message_queue_->channel_messages_.front();
+      DCHECK(m->order_number != kOutOfOrderNumber);
+      message_queue_->channel_messages_.pop_front();
+    } else {
+      // No messages to process
+      return;
+    }
+
+    if (!message_queue_->out_of_order_messages_.empty() ||
+        !message_queue_->channel_messages_.empty()) {
+      message_queue_->ScheduleHandleMessage();
+    }
+  }
+
+  bool retry_message = false;
+  stub = stubs_.get(m->message.routing_id());
+  if (stub) {
+    if (!stub->IsScheduled()) {
+      retry_message = true;
+    }
+    if (stub->IsPreempted()) {
+      retry_message = true;
+      message_queue_->ScheduleHandleMessage();
+    }
+  }
+
+  if (retry_message) {
+    base::AutoLock auto_lock(message_queue_->channel_messages_lock_);
+    if (m->order_number == kOutOfOrderNumber)
+      message_queue_->out_of_order_messages_.push_front(m);
+    else
+      message_queue_->channel_messages_.push_front(m);
     return;
+  }
 
-  IPC::Message* m = NULL;
-  GpuCommandBufferStub* stub = NULL;
+  scoped_ptr<GpuChannelMessage> scoped_message(m);
+  const uint32_t order_number = m->order_number;
+  const int32_t routing_id = m->message.routing_id();
 
-  m = deferred_messages_.front();
-  stub = stubs_.get(m->routing_id());
-  if (stub) {
-    if (!stub->IsScheduled())
-      return;
-    if (stub->IsPreempted()) {
-      OnScheduled();
+  // TODO(dyen): Temporary handling of old sync points.
+  // This must ensure that the sync point will be retired. Normally we'll
+  // find the stub based on the routing ID, and associate the sync point
+  // with it, but if that fails for any reason (channel or stub already
+  // deleted, invalid routing id), we need to retire the sync point
+  // immediately.
+  if (m->message.type() == GpuCommandBufferMsg_InsertSyncPoint::ID) {
+    const bool retire = m->retire_sync_point;
+    const uint32_t sync_point = m->sync_point_number;
+    if (stub) {
+      stub->AddSyncPoint(sync_point);
+      if (retire) {
+        m->message =
+            GpuCommandBufferMsg_RetireSyncPoint(routing_id, sync_point);
+      }
+    } else {
+      current_order_num_ = order_number;
+      gpu_channel_manager_->sync_point_manager()->RetireSyncPoint(sync_point);
+      MessageProcessed(order_number);
       return;
     }
   }
 
-  scoped_ptr<IPC::Message> message(m);
-  deferred_messages_.pop_front();
+  IPC::Message* message = &m->message;
   bool message_processed = true;
 
-  currently_processing_message_ = message.get();
-  bool result;
-  if (message->routing_id() == MSG_ROUTING_CONTROL)
+  DVLOG(1) << "received message @" << message << " on channel @" << this
+           << " with type " << message->type();
+
+  if (order_number != kOutOfOrderNumber) {
+    // Make sure this is a valid unprocessed order number.
+    DCHECK(order_number <= GetUnprocessedOrderNum() &&
+           order_number >= GetProcessedOrderNum());
+
+    current_order_num_ = order_number;
+  }
+  bool result = false;
+  if (routing_id == MSG_ROUTING_CONTROL)
     result = OnControlMessageReceived(*message);
   else
     result = router_.RouteMessage(*message);
-  currently_processing_message_ = NULL;
 
   if (!result) {
     // Respond to sync messages even if router failed to route.
     if (message->is_sync()) {
       IPC::Message* reply = IPC::SyncMessage::GenerateReply(&*message);
       reply->set_reply_error();
       Send(reply);
     }
   } else {
     // If the command buffer becomes unscheduled as a result of handling the
     // message but still has more commands to process, synthesize an IPC
     // message to flush that command buffer.
     if (stub) {
       if (stub->HasUnprocessedCommands()) {
-        deferred_messages_.push_front(new GpuCommandBufferMsg_Rescheduled(
-            stub->route_id()));
+        message_queue_->PushUnfinishedMessage(
+            order_number, GpuCommandBufferMsg_Rescheduled(stub->route_id()));
         message_processed = false;
       }
     }
   }
   if (message_processed)
-    MessageProcessed();
-
-  if (!deferred_messages_.empty()) {
-    OnScheduled();
-  }
+    MessageProcessed(order_number);
 }
 
 void GpuChannel::OnCreateOffscreenCommandBuffer(
     const gfx::Size& size,
     const GPUCreateCommandBufferConfig& init_params,
     int32 route_id,
     bool* succeeded) {
   TRACE_EVENT1("gpu", "GpuChannel::OnCreateOffscreenCommandBuffer", "route_id",
                route_id);
 
@@ skipping 48 matching lines @@
   }
 }
 
 void GpuChannel::OnCreateJpegDecoder(int32 route_id, IPC::Message* reply_msg) {
   if (!jpeg_decoder_) {
     jpeg_decoder_.reset(new GpuJpegDecodeAccelerator(this, io_task_runner_));
   }
   jpeg_decoder_->AddClient(route_id, reply_msg);
 }
 
-void GpuChannel::MessageProcessed() {
-  messages_processed_++;
+void GpuChannel::MessageProcessed(uint32_t order_number) {
+  if (order_number != kOutOfOrderNumber) {
+    DCHECK(current_order_num_ == order_number);
+    processed_order_num_ = order_number;

[jbauman, 2015/09/01 20:03:06]

Can we DCHECK that processed_order_num_ only ever increases?

[David Yen, 2015/09/01 22:46:52]

Missed this in my last patch. Done now.

+  }
   if (preempting_flag_.get()) {
     io_task_runner_->PostTask(
-        FROM_HERE, base::Bind(&GpuChannelMessageFilter::MessageProcessed,
-                              filter_, messages_processed_));
+        FROM_HERE,
+        base::Bind(&GpuChannelMessageFilter::OnMessageProcessed, filter_));
   }
 }
 
 void GpuChannel::CacheShader(const std::string& key,
                              const std::string& shader) {
   gpu_channel_manager_->Send(
       new GpuHostMsg_CacheShader(client_id_, key, shader));
 }
 
 void GpuChannel::AddFilter(IPC::MessageFilter* filter) {
@@ skipping 48 matching lines @@
                                           client_id_);
     }
   }
 }
 
 void GpuChannel::HandleUpdateValueState(
     unsigned int target, const gpu::ValueState& state) {
   pending_valuebuffer_state_->UpdateState(target, state);
 }
 
+uint32_t GpuChannel::GetUnprocessedOrderNum() const {
+  return message_queue_->GetUnprocessedOrderNum();
+}
+
 }  // namespace content