Revert of Adds TimeDomains to the TaskQueueManager

components/scheduler/base/task_queue_manager.cc

Index: components/scheduler/base/task_queue_manager.cc
diff --git a/components/scheduler/base/task_queue_manager.cc b/components/scheduler/base/task_queue_manager.cc
index 4a3602ebc29d056a545eaee0f4d68574462cacf3..2f557355deda5b264f93b01d47535009bd0ed9a2 100644
--- a/components/scheduler/base/task_queue_manager.cc
+++ b/components/scheduler/base/task_queue_manager.cc
@@ -8,11 +8,15 @@
 #include <set>
 
 #include "base/bind.h"
-#include "components/scheduler/base/real_time_domain.h"
+#include "components/scheduler/base/lazy_now.h"
 #include "components/scheduler/base/task_queue_impl.h"
 #include "components/scheduler/base/task_queue_manager_delegate.h"
 #include "components/scheduler/base/task_queue_selector.h"
 #include "components/scheduler/base/task_queue_sets.h"
+
+namespace {
+const int64_t kMaxTimeTicks = std::numeric_limits<int64>::max();
+}
 
 namespace scheduler {
 
@@ -42,10 +46,6 @@
       base::Bind(&TaskQueueManager::DoWork, weak_factory_.GetWeakPtr(), true);
   do_work_closure_ =
       base::Bind(&TaskQueueManager::DoWork, weak_factory_.GetWeakPtr(), false);
-
-  real_time_domain_ =
-      make_scoped_refptr(new RealTimeDomain(delegate.get(), do_work_closure_));
-  RegisterTimeDomain(real_time_domain_);
 }
 
 TaskQueueManager::~TaskQueueManager() {
@@ -56,16 +56,6 @@
     (*queues_.begin())->UnregisterTaskQueue();
 
   selector_.SetTaskQueueSelectorObserver(nullptr);
-}
-
-void TaskQueueManager::RegisterTimeDomain(
-    const scoped_refptr<TimeDomain>& time_domain) {
-  time_domains_.insert(time_domain);
-}
-
-void TaskQueueManager::UnregisterTimeDomain(
-    const scoped_refptr<TimeDomain>& time_domain) {
-  time_domains_.erase(time_domain);
 }
 
 scoped_refptr<internal::TaskQueueImpl> TaskQueueManager::NewTaskQueue(
@@ -73,13 +63,9 @@
   TRACE_EVENT1(tracing_category_,
                "TaskQueueManager::NewTaskQueue", "queue_name", spec.name);
   DCHECK(main_thread_checker_.CalledOnValidThread());
-  TimeDomain* time_domain =
-      spec.time_domain ? spec.time_domain : real_time_domain_.get();
-  DCHECK(time_domains_.find(make_scoped_refptr(time_domain)) !=
-         time_domains_.end());
   scoped_refptr<internal::TaskQueueImpl> queue(
       make_scoped_refptr(new internal::TaskQueueImpl(
-          this, time_domain, spec, disabled_by_default_tracing_category_,
+          this, spec, disabled_by_default_tracing_category_,
           disabled_by_default_verbose_tracing_category_)));
   queues_.insert(queue);
   selector_.AddQueue(queue.get());
@@ -105,16 +91,156 @@
   queues_to_delete_.insert(task_queue);
   queues_.erase(task_queue);
   selector_.RemoveQueue(task_queue.get());
+
+  // We need to remove |task_queue| from delayed_wakeup_multimap_ which is a
+  // little awkward since it's keyed by time. O(n) running time.
+  for (DelayedWakeupMultimap::iterator iter = delayed_wakeup_multimap_.begin();
+       iter != delayed_wakeup_multimap_.end();) {
+    if (iter->second == task_queue.get()) {
+      DelayedWakeupMultimap::iterator temp = iter;
+      iter++;
+      // O(1) amortized.
+      delayed_wakeup_multimap_.erase(temp);
+    } else {
+      iter++;
+    }
+  }
+
+  // |newly_updatable_| might contain |task_queue|, we use
+  // MoveNewlyUpdatableQueuesIntoUpdatableQueueSet to flush it out.
+  MoveNewlyUpdatableQueuesIntoUpdatableQueueSet();
+  updatable_queue_set_.erase(task_queue.get());
+}
+
+base::TimeTicks TaskQueueManager::NextPendingDelayedTaskRunTime() {
+  DCHECK(main_thread_checker_.CalledOnValidThread());
+  bool found_pending_task = false;
+  base::TimeTicks next_pending_delayed_task(
+      base::TimeTicks::FromInternalValue(kMaxTimeTicks));
+  for (auto& queue : queues_) {
+    base::TimeTicks queues_next_pending_delayed_task;
+    if (queue->NextPendingDelayedTaskRunTime(
+            &queues_next_pending_delayed_task)) {
+      found_pending_task = true;
+      next_pending_delayed_task =
+          std::min(next_pending_delayed_task, queues_next_pending_delayed_task);
+    }
+  }
+
+  if (!found_pending_task)
+    return base::TimeTicks();
+
+  DCHECK_NE(next_pending_delayed_task,
+            base::TimeTicks::FromInternalValue(kMaxTimeTicks));
+  return next_pending_delayed_task;
+}
+
+void TaskQueueManager::RegisterAsUpdatableTaskQueue(
+    internal::TaskQueueImpl* queue) {
+  base::AutoLock lock(newly_updatable_lock_);
+  newly_updatable_.push_back(queue);
+}
+
+void TaskQueueManager::UnregisterAsUpdatableTaskQueue(
+    internal::TaskQueueImpl* queue) {
+  DCHECK(main_thread_checker_.CalledOnValidThread());
+  MoveNewlyUpdatableQueuesIntoUpdatableQueueSet();
+#ifndef NDEBUG
+  {
+    base::AutoLock lock(newly_updatable_lock_);
+    DCHECK(!(updatable_queue_set_.find(queue) == updatable_queue_set_.end() &&
+             std::find(newly_updatable_.begin(), newly_updatable_.end(),
+                       queue) != newly_updatable_.end()));
+  }
+#endif
+  updatable_queue_set_.erase(queue);
+}
+
+void TaskQueueManager::MoveNewlyUpdatableQueuesIntoUpdatableQueueSet() {
+  DCHECK(main_thread_checker_.CalledOnValidThread());
+  base::AutoLock lock(newly_updatable_lock_);
+  while (!newly_updatable_.empty()) {
+    updatable_queue_set_.insert(newly_updatable_.back());
+    newly_updatable_.pop_back();
+  }
 }
 
 void TaskQueueManager::UpdateWorkQueues(
     bool should_trigger_wakeup,
     const internal::TaskQueueImpl::Task* previous_task) {
+  DCHECK(main_thread_checker_.CalledOnValidThread());
   TRACE_EVENT0(disabled_by_default_tracing_category_,
                "TaskQueueManager::UpdateWorkQueues");
-
-  for (const scoped_refptr<TimeDomain>& time_domain : time_domains_) {
-    time_domain->UpdateWorkQueues(should_trigger_wakeup, previous_task);
+  internal::LazyNow lazy_now(delegate().get());
+
+  // Move any ready delayed tasks into the incomming queues.
+  WakeupReadyDelayedQueues(&lazy_now);
+
+  MoveNewlyUpdatableQueuesIntoUpdatableQueueSet();
+
+  auto iter = updatable_queue_set_.begin();
+  while (iter != updatable_queue_set_.end()) {
+    internal::TaskQueueImpl* queue = *iter++;
+    // NOTE Update work queue may erase itself from |updatable_queue_set_|.
+    // This is fine, erasing an element won't invalidate any interator, as long
+    // as the iterator isn't the element being delated.
+    if (queue->work_queue().empty())
+      queue->UpdateWorkQueue(&lazy_now, should_trigger_wakeup, previous_task);
+  }
+}
+
+void TaskQueueManager::ScheduleDelayedWorkTask(
+    scoped_refptr<internal::TaskQueueImpl> queue,
+    base::TimeTicks delayed_run_time) {
+  internal::LazyNow lazy_now(delegate().get());
+  ScheduleDelayedWork(queue.get(), delayed_run_time, &lazy_now);
+}
+
+void TaskQueueManager::ScheduleDelayedWork(internal::TaskQueueImpl* queue,
+                                           base::TimeTicks delayed_run_time,
+                                           internal::LazyNow* lazy_now) {
+  if (!delegate_->BelongsToCurrentThread()) {
+    // NOTE posting a delayed task from a different thread is not expected to be
+    // common. This pathway is less optimal than perhaps it could be because
+    // it causes two main thread tasks to be run.  Should this assumption prove
+    // to be false in future, we may need to revisit this.
+    delegate_->PostTask(
+        FROM_HERE, base::Bind(&TaskQueueManager::ScheduleDelayedWorkTask,
+                              weak_factory_.GetWeakPtr(),
+                              scoped_refptr<internal::TaskQueueImpl>(queue),
+                              delayed_run_time));
+    return;
+  }
+
+  // Make sure there's one (and only one) task posted to |delegate_|
+  // to call |DelayedDoWork| at |delayed_run_time|.
+  if (delayed_wakeup_multimap_.find(delayed_run_time) ==
+      delayed_wakeup_multimap_.end()) {
+    base::TimeDelta delay =
+        std::max(base::TimeDelta(), delayed_run_time - lazy_now->Now());
+    delegate_->PostDelayedTask(FROM_HERE, do_work_closure_, delay);
+  }
+  delayed_wakeup_multimap_.insert(std::make_pair(delayed_run_time, queue));
+}
+
+void TaskQueueManager::WakeupReadyDelayedQueues(internal::LazyNow* lazy_now) {
+  // Wake up any queues with pending delayed work.  Note std::multipmap stores
+  // the elements sorted by key, so the begin() iterator points to the earliest
+  // queue to wakeup.
+  std::set<internal::TaskQueueImpl*> dedup_set;
+  while (!delayed_wakeup_multimap_.empty()) {
+    DelayedWakeupMultimap::iterator next_wakeup =
+        delayed_wakeup_multimap_.begin();
+    if (next_wakeup->first > lazy_now->Now())
+      break;
+    // A queue could have any number of delayed tasks pending so it's worthwhile
+    // deduping calls to MoveReadyDelayedTasksToIncomingQueue since it takes a
+    // lock.  NOTE the order in which these are called matters since the order
+    // in which EnqueueTaskLocks is called is respected when choosing which
+    // queue to execute a task from.
+    if (dedup_set.insert(next_wakeup->second).second)
+      next_wakeup->second->MoveReadyDelayedTasksToIncomingQueue(lazy_now);
+    delayed_wakeup_multimap_.erase(next_wakeup);
   }
 }
 
@@ -177,20 +303,12 @@
   // TODO(alexclarke): Consider refactoring the above loop to terminate only
   // when there's no more work left to be done, rather than posting a
   // continuation task.
-  if (!selector_.EnabledWorkQueuesEmpty() || TryAdvanceTimeDomains()) {
+  if (!selector_.EnabledWorkQueuesEmpty()) {
     MaybePostDoWorkOnMainRunner();
   } else {
     // Tell the task runner we have no more work.
     delegate_->OnNoMoreImmediateWork();
   }
-}
-
-bool TaskQueueManager::TryAdvanceTimeDomains() {
-  bool can_advance = false;
-  for (const scoped_refptr<TimeDomain>& time_domain : time_domains_) {
-    can_advance |= time_domain->MaybeAdvanceTime();
-  }
-  return can_advance;
 }
 
 bool TaskQueueManager::SelectQueueToService(
@@ -308,9 +426,9 @@
   if (should_run)
     state->SetString("selected_queue", selected_queue->GetName());
 
-  state->BeginArray("time_domains");
-  for (auto& time_domain : time_domains_)
-    time_domain->AsValueInto(state.get());
+  state->BeginArray("updatable_queue_set");
+  for (auto& queue : updatable_queue_set_)
+    state->AppendString(queue->GetName());
   state->EndArray();
   return state;
 }