(reland) Adds TimeDomains to the TaskQueueManager

components/scheduler/base/task_queue_impl.cc

 // Copyright 2015 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/base/task_queue_impl.h"
 
 #include "components/scheduler/base/task_queue_manager.h"
 #include "components/scheduler/base/task_queue_manager_delegate.h"
+#include "components/scheduler/base/time_domain.h"
 
 namespace scheduler {
 namespace internal {
 
 TaskQueueImpl::TaskQueueImpl(
     TaskQueueManager* task_queue_manager,
+    const scoped_refptr<TimeDomain>& time_domain,
     const Spec& spec,
     const char* disabled_by_default_tracing_category,
     const char* disabled_by_default_verbose_tracing_category)
     : thread_id_(base::PlatformThread::CurrentId()),
-      any_thread_(task_queue_manager, spec.pump_policy),
+      any_thread_(task_queue_manager, spec.pump_policy, time_domain),
       name_(spec.name),
       disabled_by_default_tracing_category_(
           disabled_by_default_tracing_category),
       disabled_by_default_verbose_tracing_category_(
           disabled_by_default_verbose_tracing_category),
       main_thread_only_(task_queue_manager),
       wakeup_policy_(spec.wakeup_policy),
       should_monitor_quiescence_(spec.should_monitor_quiescence),
-      should_notify_observers_(spec.should_notify_observers) {}
+      should_notify_observers_(spec.should_notify_observers) {
+  DCHECK(time_domain.get());
+}
 
 TaskQueueImpl::~TaskQueueImpl() {}
 
 TaskQueueImpl::Task::Task()
     : PendingTask(tracked_objects::Location(),
                   base::Closure(),
                   base::TimeTicks(),
                   true),
 #ifndef NDEBUG
       enqueue_order_set_(false),
 #endif
       enqueue_order_(0) {
   sequence_num = 0;
 }
 
 TaskQueueImpl::Task::Task(const tracked_objects::Location& posted_from,
                           const base::Closure& task,
                           int sequence_number,
                           bool nestable)
     : PendingTask(posted_from, task, base::TimeTicks(), nestable),
 #ifndef NDEBUG
       enqueue_order_set_(false),
 #endif
       enqueue_order_(0) {
   sequence_num = sequence_number;
 }
 
-TaskQueueImpl::AnyThread::AnyThread(TaskQueueManager* task_queue_manager,
-                                    PumpPolicy pump_policy)
-    : task_queue_manager(task_queue_manager), pump_policy(pump_policy) {}
+TaskQueueImpl::AnyThread::AnyThread(
+    TaskQueueManager* task_queue_manager,
+    PumpPolicy pump_policy,
+    const scoped_refptr<TimeDomain>& time_domain)
+    : task_queue_manager(task_queue_manager),
+      pump_policy(pump_policy),
+      time_domain(time_domain) {}
 
 TaskQueueImpl::AnyThread::~AnyThread() {}
 
 TaskQueueImpl::MainThreadOnly::MainThreadOnly(
     TaskQueueManager* task_queue_manager)
     : task_queue_manager(task_queue_manager),
       set_index(0) {}
 
 TaskQueueImpl::MainThreadOnly::~MainThreadOnly() {}
 
 void TaskQueueImpl::UnregisterTaskQueue() {
   base::AutoLock lock(any_thread_lock_);
   if (!any_thread().task_queue_manager)
     return;
+  any_thread().time_domain->UnregisterQueue(this);
+  any_thread().time_domain = nullptr;
   any_thread().task_queue_manager->UnregisterTaskQueue(this);
 
   any_thread().task_queue_manager = nullptr;
   main_thread_only().task_queue_manager = nullptr;
   any_thread().delayed_task_queue = std::priority_queue<Task>();
   any_thread().incoming_queue = std::queue<Task>();
   main_thread_only().work_queue = std::queue<Task>();
 }
 
 bool TaskQueueImpl::RunsTasksOnCurrentThread() const {
@@ skipping 14 matching lines @@
   return PostDelayedTaskImpl(from_here, task, delay, TaskType::NON_NESTABLE);
 }
 
 bool TaskQueueImpl::PostDelayedTaskAt(
     const tracked_objects::Location& from_here,
     const base::Closure& task,
     base::TimeTicks desired_run_time) {
   base::AutoLock lock(any_thread_lock_);
   if (!any_thread().task_queue_manager)
     return false;
-  LazyNow lazy_now(any_thread().task_queue_manager->delegate().get());
+  LazyNow lazy_now(any_thread().time_domain->GetLazyNow());
   return PostDelayedTaskLocked(&lazy_now, from_here, task, desired_run_time,
                                TaskType::NORMAL);
 }
 
 bool TaskQueueImpl::PostDelayedTaskImpl(
     const tracked_objects::Location& from_here,
     const base::Closure& task,
     base::TimeDelta delay,
     TaskType task_type) {
   base::AutoLock lock(any_thread_lock_);
   if (!any_thread().task_queue_manager)
     return false;
-  LazyNow lazy_now(any_thread().task_queue_manager->delegate().get());
+  LazyNow lazy_now(any_thread().time_domain->GetLazyNow());
   base::TimeTicks desired_run_time;
   if (delay > base::TimeDelta())
     desired_run_time = lazy_now.Now() + delay;
   return PostDelayedTaskLocked(&lazy_now, from_here, task, desired_run_time,
                                task_type);
 }
 
 bool TaskQueueImpl::PostDelayedTaskLocked(
     LazyNow* lazy_now,
     const tracked_objects::Location& from_here,
     const base::Closure& task,
     base::TimeTicks desired_run_time,
     TaskType task_type) {
   DCHECK(any_thread().task_queue_manager);
   Task pending_task(from_here, task,
                     any_thread().task_queue_manager->GetNextSequenceNumber(),
                     task_type != TaskType::NON_NESTABLE);
   any_thread().task_queue_manager->DidQueueTask(pending_task);
 
   if (!desired_run_time.is_null()) {
     pending_task.delayed_run_time = std::max(lazy_now->Now(), desired_run_time);
     // TODO(alexclarke): consider emplace() when C++11 library features allowed.
     any_thread().delayed_task_queue.push(pending_task);
     TraceQueueSize(true);
     // Schedule a later call to MoveReadyDelayedTasksToIncomingQueue.
-    any_thread().task_queue_manager->ScheduleDelayedWork(this, desired_run_time,
-                                                         lazy_now);
+    if (base::PlatformThread::CurrentId() == thread_id_) {
+      any_thread().time_domain->ScheduleDelayedWork(this, desired_run_time,
+                                                    lazy_now);
+    } else {
+      // TimeDomain::ScheduleDelayedWork is not thread safe, so post a
+      // non-delayed task (which is thread safe) to schedule it.

[Sami, 2015/11/19 13:01:03]

Could you retain the comment about this being a little inefficient but we don't
expect this to be common?

[alex clarke (OOO till 29th), 2015/11/19 13:54:47]

Done.

+      Task thread_hop_task(
+          FROM_HERE, base::Bind(&TaskQueueImpl::ScheduleDelayedWorkTask, this,
+                                desired_run_time),
+          any_thread().task_queue_manager->GetNextSequenceNumber(), true);

[Sami, 2015/11/19 13:01:03]

I assume we can't reuse the sequence number generated above for some reason?
Maybe add a comment?

[alex clarke (OOO till 29th), 2015/11/19 13:54:47]

Feels a bit trixy to do that :(

+      any_thread().task_queue_manager->DidQueueTask(thread_hop_task);
+      pending_task.set_enqueue_order(thread_hop_task.sequence_num);
+      EnqueueTaskLocked(thread_hop_task);
+    }
     return true;
   }
   pending_task.set_enqueue_order(pending_task.sequence_num);
   EnqueueTaskLocked(pending_task);
   return true;
 }
 
+void TaskQueueImpl::ScheduleDelayedWorkTask(base::TimeTicks desired_run_time) {
+  LazyNow lazy_now(any_thread().time_domain->GetLazyNow());
+  any_thread().time_domain->ScheduleDelayedWork(this, desired_run_time,
+                                                &lazy_now);
+}
+
 void TaskQueueImpl::MoveReadyDelayedTasksToIncomingQueue(LazyNow* lazy_now) {
   base::AutoLock lock(any_thread_lock_);
   if (!any_thread().task_queue_manager)
     return;
 
   MoveReadyDelayedTasksToIncomingQueueLocked(lazy_now);
 }
 
 void TaskQueueImpl::MoveReadyDelayedTasksToIncomingQueueLocked(
     LazyNow* lazy_now) {
@@ skipping 50 matching lines @@
   if (any_thread().pump_policy == PumpPolicy::MANUAL)
     return false;
   if (any_thread().pump_policy == PumpPolicy::AFTER_WAKEUP &&
       (!should_trigger_wakeup || TaskIsOlderThanQueuedTasks(previous_task)))
     return false;
   if (any_thread().incoming_queue.empty())
     return false;
   return true;
 }
 
-bool TaskQueueImpl::NextPendingDelayedTaskRunTime(
-    base::TimeTicks* next_pending_delayed_task) {
-  base::AutoLock lock(any_thread_lock_);
-  if (any_thread().delayed_task_queue.empty())
-    return false;
-  *next_pending_delayed_task =
-      any_thread().delayed_task_queue.top().delayed_run_time;
-  return true;
-}
-
 void TaskQueueImpl::UpdateWorkQueue(LazyNow* lazy_now,
                                     bool should_trigger_wakeup,
                                     const Task* previous_task) {
   DCHECK(main_thread_only().work_queue.empty());
   base::AutoLock lock(any_thread_lock_);
   if (!ShouldAutoPumpQueueLocked(should_trigger_wakeup, previous_task))
     return;
   MoveReadyDelayedTasksToIncomingQueueLocked(lazy_now);
   std::swap(main_thread_only().work_queue, any_thread().incoming_queue);
-  // |any_thread().incoming_queue| is now empty so
-  // TaskQueueManager::UpdateQueues no longer needs to consider
-  // this queue for reloading.
-  any_thread().task_queue_manager->UnregisterAsUpdatableTaskQueue(this);
+  // |any_thread().incoming_queue| is now empty so TimeDomain::UpdateQueues
+  // no longer needs to consider this queue for reloading.
+  any_thread().time_domain->UnregisterAsUpdatableTaskQueue(this);
   if (!main_thread_only().work_queue.empty()) {
     DCHECK(any_thread().task_queue_manager);
     any_thread().task_queue_manager->selector_.GetTaskQueueSets()->OnPushQueue(
         this);
     TraceQueueSize(true);
   }
 }
 
 TaskQueueImpl::Task TaskQueueImpl::TakeTaskFromWorkQueue() {
   // TODO(alexclarke): consider std::move() when allowed.
@@ skipping 22 matching lines @@
                      main_thread_only().work_queue.size() +
                      any_thread().delayed_task_queue.size());
   if (!is_locked)
     any_thread_lock_.Release();
 }
 
 void TaskQueueImpl::EnqueueTaskLocked(const Task& pending_task) {
   if (!any_thread().task_queue_manager)
     return;
   if (any_thread().incoming_queue.empty())
-    any_thread().task_queue_manager->RegisterAsUpdatableTaskQueue(this);
+    any_thread().time_domain->RegisterAsUpdatableTaskQueue(this);
   if (any_thread().pump_policy == PumpPolicy::AUTO &&
       any_thread().incoming_queue.empty()) {
     any_thread().task_queue_manager->MaybePostDoWorkOnMainRunner();
   }
   // TODO(alexclarke): consider std::move() when allowed.
   any_thread().incoming_queue.push(pending_task);
   TraceQueueSize(true);
 }
 
 void TaskQueueImpl::EnqueueDelayedTaskLocked(const Task& pending_task) {
   if (!any_thread().task_queue_manager)
     return;
   if (any_thread().incoming_queue.empty())
-    any_thread().task_queue_manager->RegisterAsUpdatableTaskQueue(this);
+    any_thread().time_domain->RegisterAsUpdatableTaskQueue(this);
+  if (any_thread().pump_policy == PumpPolicy::AUTO &&
+      any_thread().incoming_queue.empty()) {
+    any_thread().task_queue_manager->MaybePostDoWorkOnMainRunner();
+  }
   // TODO(alexclarke): consider std::move() when allowed.
   any_thread().incoming_queue.push(pending_task);
   any_thread().incoming_queue.back().set_enqueue_order(
       any_thread().task_queue_manager->GetNextSequenceNumber());
   TraceQueueSize(true);
 }
 
 void TaskQueueImpl::SetPumpPolicy(PumpPolicy pump_policy) {
   base::AutoLock lock(any_thread_lock_);
   if (pump_policy == PumpPolicy::AUTO &&
       any_thread().pump_policy != PumpPolicy::AUTO) {
     PumpQueueLocked();
   }
   any_thread().pump_policy = pump_policy;
 }
 
 void TaskQueueImpl::PumpQueueLocked() {
   if (!any_thread().task_queue_manager)
     return;
 
-  LazyNow lazy_now(any_thread().task_queue_manager->delegate().get());
+  LazyNow lazy_now(any_thread().time_domain->GetLazyNow());
   MoveReadyDelayedTasksToIncomingQueueLocked(&lazy_now);
 
   bool was_empty = main_thread_only().work_queue.empty();
   while (!any_thread().incoming_queue.empty()) {
     // TODO(alexclarke): consider std::move() when allowed.
     main_thread_only().work_queue.push(any_thread().incoming_queue.front());
     any_thread().incoming_queue.pop();
   }
-  // |incoming_queue| is now empty so TaskQueueManager::UpdateQueues no longer
-  // needs to consider this queue for reloading.
-  any_thread().task_queue_manager->UnregisterAsUpdatableTaskQueue(this);
+  // |incoming_queue| is now empty so TimeDomain::UpdateQueues no longer needs
+  // to consider this queue for reloading.
+  any_thread().time_domain->UnregisterAsUpdatableTaskQueue(this);
   if (!main_thread_only().work_queue.empty()) {
     if (was_empty) {
       any_thread()
           .task_queue_manager->selector_.GetTaskQueueSets()
           ->OnPushQueue(this);
     }
     any_thread().task_queue_manager->MaybePostDoWorkOnMainRunner();
   }
 }
 
@@ skipping 76 matching lines @@
     default:
       NOTREACHED();
       return nullptr;
   }
 }
 
 void TaskQueueImpl::AsValueInto(base::trace_event::TracedValue* state) const {
   base::AutoLock lock(any_thread_lock_);
   state->BeginDictionary();
   state->SetString("name", GetName());
+  state->SetString("time_domain_name", any_thread().time_domain->GetName());
   state->SetString("pump_policy", PumpPolicyToString(any_thread().pump_policy));
   state->SetString("wakeup_policy", 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", any_thread().incoming_queue.size());
   state->SetInteger("work_queue_size", main_thread_only().work_queue.size());
   state->SetInteger("delayed_task_queue_size",
                     any_thread().delayed_task_queue.size());
   if (verbose_tracing_enabled) {
@@ skipping 31 matching lines @@
 }
 
 void TaskQueueImpl::NotifyDidProcessTask(
     const base::PendingTask& pending_task) {
   DCHECK(should_notify_observers_);
   FOR_EACH_OBSERVER(base::MessageLoop::TaskObserver,
                     main_thread_only().task_observers,
                     DidProcessTask(pending_task));
 }
 
+void TaskQueueImpl::SetTimeDomain(
+    const scoped_refptr<TimeDomain>& time_domain) {
+  base::AutoLock lock(any_thread_lock_);

[Sami, 2015/11/19 13:01:03]

Looks like this should be main thread only since TimeDomain::ScheduleDelayedWork
is?

[alex clarke (OOO till 29th), 2015/11/19 13:54:47]

Done, MigrateQueue is main thread only.

+  if (time_domain == any_thread().time_domain)
+    return;
+
+  any_thread().time_domain->MigrateQueue(this, time_domain.get());
+  any_thread().time_domain = time_domain;
+}
+
 // static
 void TaskQueueImpl::QueueAsValueInto(const std::queue<Task>& queue,
                                      base::trace_event::TracedValue* state) {
   std::queue<Task> queue_copy(queue);
   while (!queue_copy.empty()) {
     TaskAsValueInto(queue_copy.front(), state);
     queue_copy.pop();
   }
 }
 
@@ skipping 15 matching lines @@
   state->SetInteger("enqueue_order", task.enqueue_order());
   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();
 }
 
+size_t TaskQueueImpl::IncomingQueueSizeForTest() const {
+  base::AutoLock lock(any_thread_lock_);
+  return any_thread().incoming_queue.size();
+}
+
 }  // namespace internal
 }  // namespace scheduler