Implement PostDelayedTaskAt for guaranteed timer ordering

components/scheduler/child/task_queue_manager.cc

 // Copyright 2014 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 "components/scheduler/child/task_queue_manager.h"
 
 #include <queue>
 #include <set>
 
 #include "base/bind.h"
 #include "base/time/default_tick_clock.h"
 #include "base/trace_event/trace_event.h"
 #include "base/trace_event/trace_event_argument.h"
 #include "components/scheduler/child/nestable_single_thread_task_runner.h"
+#include "components/scheduler/child/task_queue.h"
 #include "components/scheduler/child/task_queue_selector.h"
 
 namespace {
 const int64_t kMaxTimeTicks = std::numeric_limits<int64>::max();
 }
 
 namespace scheduler {
 namespace internal {
 
 // Now() is somewhat expensive so it makes sense not to call Now() unless we
@@ skipping 12 matching lines @@
     if (now_.is_null())
       now_ = task_queue_manager_->Now();
     return now_;
   }
 
  private:
   TaskQueueManager* task_queue_manager_;  // NOT OWNED
   base::TimeTicks now_;
 };
 
-class TaskQueue : public base::SingleThreadTaskRunner {
+class TaskQueueImpl : public TaskQueue {
  public:
-  TaskQueue(TaskQueueManager* task_queue_manager,
-            const char* disabled_by_default_tracing_category,
-            const char* disabled_by_default_verbose_tracing_category);
+  TaskQueueImpl(TaskQueueManager* task_queue_manager,
+                const char* disabled_by_default_tracing_category,
+                const char* disabled_by_default_verbose_tracing_category);
 
-  // base::SingleThreadTaskRunner implementation.
+  // TaskQueue :implementation.
   bool RunsTasksOnCurrentThread() const override;
   bool PostDelayedTask(const tracked_objects::Location& from_here,
                        const base::Closure& task,
                        base::TimeDelta delay) override {
-    return PostDelayedTaskImpl(from_here, task, delay, TaskType::NORMAL);
+    return PostDelayedTaskImpl(from_here, task, delay, base::TimeTicks(),
+                               TaskType::NORMAL);
   }
 
   bool PostNonNestableDelayedTask(const tracked_objects::Location& from_here,
                                   const base::Closure& task,
                                   base::TimeDelta delay) override {
-    return PostDelayedTaskImpl(from_here, task, delay, TaskType::NON_NESTABLE);
+    return PostDelayedTaskImpl(from_here, task, delay, base::TimeTicks(),
+                               TaskType::NON_NESTABLE);
+  }
+  bool PostDelayedTaskAt(const tracked_objects::Location& from_here,
+                         const base::Closure& task,
+                         base::TimeTicks desired_run_time) override {
+    return PostDelayedTaskImpl(from_here, task, base::TimeDelta(),
+                               desired_run_time, TaskType::NORMAL);
   }
 
   TaskQueueManager::QueueState GetQueueState() const;
 
   void SetPumpPolicy(TaskQueueManager::PumpPolicy pump_policy);
   void PumpQueue();
 
   bool NextPendingDelayedTaskRunTime(
       base::TimeTicks* next_pending_delayed_task);
 
@@ skipping 19 matching lines @@
   }
 
   void AsValueInto(base::trace_event::TracedValue* state) const;
 
  private:
   enum class TaskType {
     NORMAL,
     NON_NESTABLE,
   };
 
-  ~TaskQueue() override;
+  ~TaskQueueImpl() override;
 
   bool PostDelayedTaskImpl(const tracked_objects::Location& from_here,
                            const base::Closure& task,
                            base::TimeDelta delay,

[alex clarke (OOO till 29th), 2015/07/16 09:29:00]

It feels a bit weird having both a delay and a desired_run_time.

How about passing in LazyNow and desired_run_time, and have PostDelayedTask &
PostNonNestableDelayedTask do the computation for desired_run_time?

If desired_run_time was null then PostDelayedTaskImpl would enqueue the task
immediately.

[Sami, 2015/07/16 12:19:32]

Yep I struggled with this interface a bit. The locking makes it a little
inconvenient, but let me give this another shot.

+                           base::TimeTicks desired_run_time,
                            TaskType task_type);
 
   // Delayed task posted to the underlying run loop, which locks |lock_| and
   // calls MoveReadyDelayedTasksToIncomingQueueLocked to process dealyed tasks
   // that need to be run now.
   void MoveReadyDelayedTasksToIncomingQueue();
 
   // Enqueues any delayed tasks which should be run now on the incoming_queue_
   // and calls ScheduleDelayedWorkLocked to ensure future tasks are scheduled.
   // Must be called with |lock_| locked.
@@ skipping 23 matching lines @@
   // This lock protects all members except the work queue, the
   // main_thread_checker_ and wakeup_policy_.
   mutable base::Lock lock_;
   base::PlatformThreadId thread_id_;
   TaskQueueManager* task_queue_manager_;
   base::TaskQueue incoming_queue_;
   TaskQueueManager::PumpPolicy pump_policy_;
   const char* name_;
   const char* disabled_by_default_tracing_category_;
   const char* disabled_by_default_verbose_tracing_category_;
+
+  // Queue-local task sequence number for maintaining the order of delayed
+  // tasks which are posted for the exact same time. Note that this will be
+  // replaced by the global sequence number when the delay has elapsed.
+  int delayed_task_sequence_number_;
   base::DelayedTaskQueue delayed_task_queue_;
   std::set<base::TimeTicks> in_flight_kick_delayed_tasks_;
 
   base::ThreadChecker main_thread_checker_;
   base::TaskQueue work_queue_;
   TaskQueueManager::WakeupPolicy wakeup_policy_;
 
-  DISALLOW_COPY_AND_ASSIGN(TaskQueue);
+  DISALLOW_COPY_AND_ASSIGN(TaskQueueImpl);
 };
 
-TaskQueue::TaskQueue(TaskQueueManager* task_queue_manager,
-                     const char* disabled_by_default_tracing_category,
-                     const char* disabled_by_default_verbose_tracing_category)
+TaskQueueImpl::TaskQueueImpl(
+    TaskQueueManager* task_queue_manager,
+    const char* disabled_by_default_tracing_category,
+    const char* disabled_by_default_verbose_tracing_category)
     : thread_id_(base::PlatformThread::CurrentId()),
       task_queue_manager_(task_queue_manager),
       pump_policy_(TaskQueueManager::PumpPolicy::AUTO),
       name_(nullptr),
       disabled_by_default_tracing_category_(
           disabled_by_default_tracing_category),
       disabled_by_default_verbose_tracing_category_(
           disabled_by_default_verbose_tracing_category),
+      delayed_task_sequence_number_(0),
       wakeup_policy_(TaskQueueManager::WakeupPolicy::CAN_WAKE_OTHER_QUEUES) {
 }
 
-TaskQueue::~TaskQueue() {
+TaskQueueImpl::~TaskQueueImpl() {
 }
 
-void TaskQueue::WillDeleteTaskQueueManager() {
+void TaskQueueImpl::WillDeleteTaskQueueManager() {
   base::AutoLock lock(lock_);
   task_queue_manager_ = nullptr;
   delayed_task_queue_ = base::DelayedTaskQueue();
   incoming_queue_ = base::TaskQueue();
   work_queue_ = base::TaskQueue();
 }
 
-bool TaskQueue::RunsTasksOnCurrentThread() const {
+bool TaskQueueImpl::RunsTasksOnCurrentThread() const {
   base::AutoLock lock(lock_);
   return base::PlatformThread::CurrentId() == thread_id_;
 }
 
-bool TaskQueue::PostDelayedTaskImpl(const tracked_objects::Location& from_here,
-                                    const base::Closure& task,
-                                    base::TimeDelta delay,
-                                    TaskType task_type) {
+bool TaskQueueImpl::PostDelayedTaskImpl(
+    const tracked_objects::Location& from_here,
+    const base::Closure& task,
+    base::TimeDelta delay,
+    base::TimeTicks desired_run_time,
+    TaskType task_type) {
   base::AutoLock lock(lock_);
   if (!task_queue_manager_)
     return false;
 
   base::PendingTask pending_task(from_here, task, base::TimeTicks(),
                                  task_type != TaskType::NON_NESTABLE);
   task_queue_manager_->DidQueueTask(pending_task);
 
-  if (delay > base::TimeDelta()) {
+  if (delay > base::TimeDelta() || !desired_run_time.is_null()) {
     base::TimeTicks now = task_queue_manager_->Now();
-    pending_task.delayed_run_time = now + delay;
+    if (!desired_run_time.is_null()) {
+      pending_task.delayed_run_time = std::max(now, desired_run_time);

[alex clarke (OOO till 29th), 2015/07/16 09:29:00]

Should we just enqueue the task if desired runtime is in the past?

[Sami, 2015/07/16 12:19:32]

I think that might change the ordering. For example let's say it's t=0 and we
get two tasks targeting t=1. If we noticed that it's already t=2 when the second
task comes in, it would skip the delayed task queue and run ahead of the first
task.

+    } else {
+      pending_task.delayed_run_time = now + delay;
+    }
+    pending_task.sequence_num = delayed_task_sequence_number_++;
     delayed_task_queue_.push(pending_task);
     TraceQueueSize(true);
     // If we changed the topmost task, then it is time to reschedule.
     if (delayed_task_queue_.top().task.Equals(pending_task.task)) {
       LazyNow lazy_now(now);
       ScheduleDelayedWorkLocked(&lazy_now);
     }
     return true;
   }
   EnqueueTaskLocked(pending_task);
   return true;
 }
 
-void TaskQueue::MoveReadyDelayedTasksToIncomingQueue() {
+void TaskQueueImpl::MoveReadyDelayedTasksToIncomingQueue() {
   DCHECK(main_thread_checker_.CalledOnValidThread());
   base::AutoLock lock(lock_);
   if (!task_queue_manager_)
     return;
 
   LazyNow lazy_now(task_queue_manager_);
   MoveReadyDelayedTasksToIncomingQueueLocked(&lazy_now);
 }
 
-void TaskQueue::MoveReadyDelayedTasksToIncomingQueueLocked(LazyNow* lazy_now) {
+void TaskQueueImpl::MoveReadyDelayedTasksToIncomingQueueLocked(
+    LazyNow* lazy_now) {
   lock_.AssertAcquired();
   // Enqueue all delayed tasks that should be running now.
   while (!delayed_task_queue_.empty() &&
          delayed_task_queue_.top().delayed_run_time <= lazy_now->Now()) {
     in_flight_kick_delayed_tasks_.erase(
         delayed_task_queue_.top().delayed_run_time);
     EnqueueTaskLocked(delayed_task_queue_.top());
     delayed_task_queue_.pop();
   }
   TraceQueueSize(true);
   ScheduleDelayedWorkLocked(lazy_now);
 }
 
-void TaskQueue::ScheduleDelayedWorkLocked(LazyNow* lazy_now) {
+void TaskQueueImpl::ScheduleDelayedWorkLocked(LazyNow* lazy_now) {
   lock_.AssertAcquired();
   // Any remaining tasks are in the future, so queue a task to kick them.
   if (!delayed_task_queue_.empty()) {
     base::TimeTicks next_run_time = delayed_task_queue_.top().delayed_run_time;
-    DCHECK_GT(next_run_time, lazy_now->Now());
+    DCHECK_GE(next_run_time, lazy_now->Now());
     // Make sure we don't have more than one
     // MoveReadyDelayedTasksToIncomingQueue posted for a particular scheduled
     // run time (note it's fine to have multiple ones in flight for distinct
     // run times).
     if (in_flight_kick_delayed_tasks_.find(next_run_time) ==
         in_flight_kick_delayed_tasks_.end()) {
       in_flight_kick_delayed_tasks_.insert(next_run_time);
       base::TimeDelta delay = next_run_time - lazy_now->Now();
       task_queue_manager_->PostDelayedTask(
           FROM_HERE,
-          Bind(&TaskQueue::MoveReadyDelayedTasksToIncomingQueue, this), delay);
+          Bind(&TaskQueueImpl::MoveReadyDelayedTasksToIncomingQueue, this),
+          delay);
     }
   }
 }
 
-TaskQueueManager::QueueState TaskQueue::GetQueueState() const {
+TaskQueueManager::QueueState TaskQueueImpl::GetQueueState() const {
   DCHECK(main_thread_checker_.CalledOnValidThread());
   if (!work_queue_.empty())
     return TaskQueueManager::QueueState::HAS_WORK;
 
   {
     base::AutoLock lock(lock_);
     if (incoming_queue_.empty()) {
       return TaskQueueManager::QueueState::EMPTY;
     } else {
       return TaskQueueManager::QueueState::NEEDS_PUMPING;
     }
   }
 }
 
-bool TaskQueue::TaskIsOlderThanQueuedTasks(const base::PendingTask* task) {
+bool TaskQueueImpl::TaskIsOlderThanQueuedTasks(const base::PendingTask* task) {
   lock_.AssertAcquired();
   // A null task is passed when UpdateQueue is called before any task is run.
   // In this case we don't want to pump an after_wakeup queue, so return true
   // here.
   if (!task)
     return true;
 
   // Return false if there are no task in the incoming queue.
   if (incoming_queue_.empty())
     return false;
 
   base::PendingTask oldest_queued_task = incoming_queue_.front();
   DCHECK(oldest_queued_task.delayed_run_time.is_null());
   DCHECK(task->delayed_run_time.is_null());
 
   // Note: the comparison is correct due to the fact that the PendingTask
   // operator inverts its comparison operation in order to work well in a heap
   // based priority queue.
   return oldest_queued_task < *task;
 }
 
-bool TaskQueue::ShouldAutoPumpQueueLocked(
+bool TaskQueueImpl::ShouldAutoPumpQueueLocked(
     bool should_trigger_wakeup,
     const base::PendingTask* previous_task) {
   lock_.AssertAcquired();
   if (pump_policy_ == TaskQueueManager::PumpPolicy::MANUAL)
     return false;
   if (pump_policy_ == TaskQueueManager::PumpPolicy::AFTER_WAKEUP &&
       (!should_trigger_wakeup || TaskIsOlderThanQueuedTasks(previous_task)))
     return false;
   if (incoming_queue_.empty())
     return false;
   return true;
 }
 
-bool TaskQueue::NextPendingDelayedTaskRunTime(
+bool TaskQueueImpl::NextPendingDelayedTaskRunTime(
     base::TimeTicks* next_pending_delayed_task) {
   base::AutoLock lock(lock_);
   if (delayed_task_queue_.empty())
     return false;
   *next_pending_delayed_task = delayed_task_queue_.top().delayed_run_time;
   return true;
 }
 
-bool TaskQueue::UpdateWorkQueue(LazyNow* lazy_now,
-                                bool should_trigger_wakeup,
-                                const base::PendingTask* previous_task) {
+bool TaskQueueImpl::UpdateWorkQueue(LazyNow* lazy_now,
+                                    bool should_trigger_wakeup,
+                                    const base::PendingTask* previous_task) {
   if (!work_queue_.empty())
     return true;
 
   {
     base::AutoLock lock(lock_);
     if (!ShouldAutoPumpQueueLocked(should_trigger_wakeup, previous_task))
       return false;
     MoveReadyDelayedTasksToIncomingQueueLocked(lazy_now);
     work_queue_.Swap(&incoming_queue_);
     TraceQueueSize(true);
     return true;
   }
 }
 
-base::PendingTask TaskQueue::TakeTaskFromWorkQueue() {
+base::PendingTask TaskQueueImpl::TakeTaskFromWorkQueue() {
   base::PendingTask pending_task = work_queue_.front();
   work_queue_.pop();
   TraceQueueSize(false);
   return pending_task;
 }
 
-void TaskQueue::TraceQueueSize(bool is_locked) const {
+void TaskQueueImpl::TraceQueueSize(bool is_locked) const {
   bool is_tracing;
   TRACE_EVENT_CATEGORY_GROUP_ENABLED(disabled_by_default_tracing_category_,
                                      &is_tracing);
   if (!is_tracing || !name_)
     return;
   if (!is_locked)
     lock_.Acquire();
   else
     lock_.AssertAcquired();
   TRACE_COUNTER1(
       disabled_by_default_tracing_category_, name_,
       incoming_queue_.size() + work_queue_.size() + delayed_task_queue_.size());
   if (!is_locked)
     lock_.Release();
 }
 
-void TaskQueue::EnqueueTaskLocked(const base::PendingTask& pending_task) {
+void TaskQueueImpl::EnqueueTaskLocked(const base::PendingTask& pending_task) {
   lock_.AssertAcquired();
   if (!task_queue_manager_)
     return;
   if (pump_policy_ == TaskQueueManager::PumpPolicy::AUTO &&
       incoming_queue_.empty())
     task_queue_manager_->MaybePostDoWorkOnMainRunner();
   incoming_queue_.push(pending_task);
   incoming_queue_.back().sequence_num =
       task_queue_manager_->GetNextSequenceNumber();
 
   if (!pending_task.delayed_run_time.is_null()) {
     // Clear the delayed run time because we've already applied the delay
     // before getting here.
     incoming_queue_.back().delayed_run_time = base::TimeTicks();
   }
   TraceQueueSize(true);
 }
 
-void TaskQueue::SetPumpPolicy(TaskQueueManager::PumpPolicy pump_policy) {
+void TaskQueueImpl::SetPumpPolicy(TaskQueueManager::PumpPolicy pump_policy) {
   base::AutoLock lock(lock_);
   if (pump_policy == TaskQueueManager::PumpPolicy::AUTO &&
       pump_policy_ != TaskQueueManager::PumpPolicy::AUTO) {
     PumpQueueLocked();
   }
   pump_policy_ = pump_policy;
 }
 
-void TaskQueue::PumpQueueLocked() {
+void TaskQueueImpl::PumpQueueLocked() {
   lock_.AssertAcquired();
   if (task_queue_manager_) {
     LazyNow lazy_now(task_queue_manager_);
     MoveReadyDelayedTasksToIncomingQueueLocked(&lazy_now);
   }
   while (!incoming_queue_.empty()) {
     work_queue_.push(incoming_queue_.front());
     incoming_queue_.pop();
   }
   if (!work_queue_.empty())
     task_queue_manager_->MaybePostDoWorkOnMainRunner();
 }
 
-void TaskQueue::PumpQueue() {
+void TaskQueueImpl::PumpQueue() {
   base::AutoLock lock(lock_);
   PumpQueueLocked();
 }
 
-void TaskQueue::AsValueInto(base::trace_event::TracedValue* state) const {
+void TaskQueueImpl::AsValueInto(base::trace_event::TracedValue* state) const {
   base::AutoLock lock(lock_);
   state->BeginDictionary();
   if (name_)
     state->SetString("name", name_);
   state->SetString("pump_policy",
                    TaskQueueManager::PumpPolicyToString(pump_policy_));
   state->SetString("wakeup_policy",
                    TaskQueueManager::WakeupPolicyToString(wakeup_policy_));
   bool verbose_tracing_enabled = false;
   TRACE_EVENT_CATEGORY_GROUP_ENABLED(
       disabled_by_default_verbose_tracing_category_, &verbose_tracing_enabled);
   state->SetInteger("incoming_queue_size", incoming_queue_.size());
   state->SetInteger("work_queue_size", work_queue_.size());
   state->SetInteger("delayed_task_queue_size", delayed_task_queue_.size());
   if (verbose_tracing_enabled) {
     state->BeginArray("incoming_queue");
     QueueAsValueInto(incoming_queue_, state);
     state->EndArray();
     state->BeginArray("work_queue");
     QueueAsValueInto(work_queue_, state);
     state->EndArray();
     state->BeginArray("delayed_task_queue");
     QueueAsValueInto(delayed_task_queue_, state);
     state->EndArray();
   }
   state->EndDictionary();
 }
 
 // static
-void TaskQueue::QueueAsValueInto(const base::TaskQueue& queue,
-                                 base::trace_event::TracedValue* state) {
+void TaskQueueImpl::QueueAsValueInto(const base::TaskQueue& queue,
+                                     base::trace_event::TracedValue* state) {
   base::TaskQueue queue_copy(queue);
   while (!queue_copy.empty()) {
     TaskAsValueInto(queue_copy.front(), state);
     queue_copy.pop();
   }
 }
 
 // static
-void TaskQueue::QueueAsValueInto(const base::DelayedTaskQueue& queue,
-                                 base::trace_event::TracedValue* state) {
+void TaskQueueImpl::QueueAsValueInto(const base::DelayedTaskQueue& queue,
+                                     base::trace_event::TracedValue* state) {
   base::DelayedTaskQueue queue_copy(queue);
   while (!queue_copy.empty()) {
     TaskAsValueInto(queue_copy.top(), state);
     queue_copy.pop();
   }
 }
 
 // static
-void TaskQueue::TaskAsValueInto(const base::PendingTask& task,
-                                base::trace_event::TracedValue* state) {
+void TaskQueueImpl::TaskAsValueInto(const base::PendingTask& task,
+                                    base::trace_event::TracedValue* state) {
   state->BeginDictionary();
   state->SetString("posted_from", task.posted_from.ToString());
   state->SetInteger("sequence_num", task.sequence_num);
   state->SetBoolean("nestable", task.nestable);
   state->SetBoolean("is_high_res", task.is_high_res);
   state->SetDouble(
       "delayed_run_time",
       (task.delayed_run_time - base::TimeTicks()).InMicroseconds() / 1000.0L);
   state->EndDictionary();
 }
@@ skipping 16 matching lines @@
           disabled_by_default_tracing_category),
       deletion_sentinel_(new DeletionSentinel()),
       weak_factory_(this) {
   DCHECK(main_task_runner->RunsTasksOnCurrentThread());
   DCHECK_LE(task_queue_count, sizeof(task_was_run_bitmap_) * CHAR_BIT)
       << "You need a bigger int for task_was_run_bitmap_";
   TRACE_EVENT_OBJECT_CREATED_WITH_ID(disabled_by_default_tracing_category,
                                      "TaskQueueManager", this);
 
   for (size_t i = 0; i < task_queue_count; i++) {
-    scoped_refptr<internal::TaskQueue> queue(make_scoped_refptr(
-        new internal::TaskQueue(this,
-                                disabled_by_default_tracing_category,
-                                disabled_by_default_verbose_tracing_category)));
+    scoped_refptr<internal::TaskQueueImpl> queue(
+        make_scoped_refptr(new internal::TaskQueueImpl(
+            this, disabled_by_default_tracing_category,
+            disabled_by_default_verbose_tracing_category)));
     queues_.push_back(queue);
   }
 
   std::vector<const base::TaskQueue*> work_queues;
   for (const auto& queue : queues_)
     work_queues.push_back(&queue->work_queue());
   selector_->RegisterWorkQueues(work_queues);
   selector_->SetTaskQueueSelectorObserver(this);
 
   do_work_from_main_thread_closure_ =
       base::Bind(&TaskQueueManager::DoWork, weak_factory_.GetWeakPtr(), true);
   do_work_from_other_thread_closure_ =
       base::Bind(&TaskQueueManager::DoWork, weak_factory_.GetWeakPtr(), false);
 }
 
 TaskQueueManager::~TaskQueueManager() {
   TRACE_EVENT_OBJECT_DELETED_WITH_ID(disabled_by_default_tracing_category_,
                                      "TaskQueueManager", this);
   for (auto& queue : queues_)
     queue->WillDeleteTaskQueueManager();
   selector_->SetTaskQueueSelectorObserver(nullptr);
 }
 
-internal::TaskQueue* TaskQueueManager::Queue(size_t queue_index) const {
+internal::TaskQueueImpl* TaskQueueManager::Queue(size_t queue_index) const {
   DCHECK_LT(queue_index, queues_.size());
   return queues_[queue_index].get();
 }
 
-scoped_refptr<base::SingleThreadTaskRunner>
-TaskQueueManager::TaskRunnerForQueue(size_t queue_index) const {
+scoped_refptr<TaskQueue> TaskQueueManager::TaskRunnerForQueue(
+    size_t queue_index) const {
   return Queue(queue_index);
 }
 
 bool TaskQueueManager::IsQueueEmpty(size_t queue_index) const {
   return Queue(queue_index)->GetQueueState() == QueueState::EMPTY;
 }
 
 TaskQueueManager::QueueState TaskQueueManager::GetQueueState(size_t queue_index)
     const {
   return Queue(queue_index)->GetQueueState();
@@ skipping 18 matching lines @@
     return base::TimeTicks();
 
   DCHECK_NE(next_pending_delayed_task,
             base::TimeTicks::FromInternalValue(kMaxTimeTicks));
   return next_pending_delayed_task;
 }
 
 void TaskQueueManager::SetPumpPolicy(size_t queue_index,
                                      PumpPolicy pump_policy) {
   DCHECK(main_thread_checker_.CalledOnValidThread());
-  internal::TaskQueue* queue = Queue(queue_index);
+  internal::TaskQueueImpl* queue = Queue(queue_index);
   queue->SetPumpPolicy(pump_policy);
 }
 
 void TaskQueueManager::SetWakeupPolicy(size_t queue_index,
                                        WakeupPolicy wakeup_policy) {
   DCHECK(main_thread_checker_.CalledOnValidThread());
-  internal::TaskQueue* queue = Queue(queue_index);
+  internal::TaskQueueImpl* queue = Queue(queue_index);
   queue->set_wakeup_policy(wakeup_policy);
 }
 
 void TaskQueueManager::PumpQueue(size_t queue_index) {
   DCHECK(main_thread_checker_.CalledOnValidThread());
-  internal::TaskQueue* queue = Queue(queue_index);
+  internal::TaskQueueImpl* queue = Queue(queue_index);
   queue->PumpQueue();
 }
 
 bool TaskQueueManager::UpdateWorkQueues(
     bool should_trigger_wakeup,
     const base::PendingTask* previous_task) {
   // TODO(skyostil): This is not efficient when the number of queues grows very
   // large due to the number of locks taken. Consider optimizing when we get
   // there.
   DCHECK(main_thread_checker_.CalledOnValidThread());
@@ skipping 69 matching lines @@
 void TaskQueueManager::DidQueueTask(const base::PendingTask& pending_task) {
   task_annotator_.DidQueueTask("TaskQueueManager::PostTask", pending_task);
 }
 
 bool TaskQueueManager::ProcessTaskFromWorkQueue(
     size_t queue_index,
     bool has_previous_task,
     base::PendingTask* previous_task) {
   DCHECK(main_thread_checker_.CalledOnValidThread());
   scoped_refptr<DeletionSentinel> protect(deletion_sentinel_);
-  internal::TaskQueue* queue = Queue(queue_index);
+  internal::TaskQueueImpl* queue = Queue(queue_index);
   base::PendingTask pending_task = queue->TakeTaskFromWorkQueue();
   task_was_run_bitmap_ |= UINT64_C(1) << queue_index;
   if (!pending_task.nestable && main_task_runner_->IsNested()) {
     // Defer non-nestable work to the main task runner.  NOTE these tasks can be
     // arbitrarily delayed so the additional delay should not be a problem.
     main_task_runner_->PostNonNestableTask(pending_task.posted_from,
                                            pending_task.task);
   } else {
     // Suppress "will" task observer notifications for the first and "did"
     // notifications for the last task in the batch to avoid duplicate
@@ skipping 19 matching lines @@
 }
 
 bool TaskQueueManager::RunsTasksOnCurrentThread() const {
   return main_task_runner_->RunsTasksOnCurrentThread();
 }
 
 bool TaskQueueManager::PostDelayedTask(
     const tracked_objects::Location& from_here,
     const base::Closure& task,
     base::TimeDelta delay) {
-  DCHECK(delay > base::TimeDelta());
+  DCHECK_GE(delay, base::TimeDelta());
   return main_task_runner_->PostDelayedTask(from_here, task, delay);
 }
 
 void TaskQueueManager::SetQueueName(size_t queue_index, const char* name) {
   DCHECK(main_thread_checker_.CalledOnValidThread());
-  internal::TaskQueue* queue = Queue(queue_index);
+  internal::TaskQueueImpl* queue = Queue(queue_index);
   queue->set_name(name);
 }
 
 void TaskQueueManager::SetWorkBatchSize(int work_batch_size) {
   DCHECK(main_thread_checker_.CalledOnValidThread());
   DCHECK_GE(work_batch_size, 1);
   work_batch_size_ = work_batch_size;
 }
 
 void TaskQueueManager::AddTaskObserver(
@@ skipping 77 matching lines @@
       return nullptr;
   }
 }
 
 void TaskQueueManager::OnTaskQueueEnabled() {
   DCHECK(main_thread_checker_.CalledOnValidThread());
   MaybePostDoWorkOnMainRunner();
 }
 
 }  // namespace scheduler