Re-land: cc: Remove WorkerPool class and instead use TaskGraphRunner directly.

cc/resources/task_graph_runner.cc

 // Copyright 2013 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 "cc/resources/worker_pool.h"
+#include "cc/resources/task_graph_runner.h"
 
 #include <algorithm>
 
-#include "base/bind.h"
 #include "base/containers/hash_tables.h"
 #include "base/debug/trace_event.h"
-#include "base/lazy_instance.h"
-#include "base/memory/linked_ptr.h"
 #include "base/strings/stringprintf.h"
-#include "base/synchronization/condition_variable.h"
-#include "base/threading/simple_thread.h"
 #include "base/threading/thread_restrictions.h"
-#include "cc/base/scoped_ptr_deque.h"
 
 namespace cc {
+namespace internal {
 
-namespace {
+Task::Task()
+    : did_schedule_(false),
+      did_run_(false) {
+}
 
-// TaskGraphRunners can process task graphs from multiple
-// workerpool instances. All members are guarded by |lock_|.
-class TaskGraphRunner : public base::DelegateSimpleThread::Delegate {
- public:
-  typedef WorkerPool::TaskGraph TaskGraph;
-  typedef WorkerPool::TaskVector TaskVector;
+Task::~Task() {
+  DCHECK(!did_run_ || did_schedule_);
+}
 
-  TaskGraphRunner(size_t num_threads, const std::string& thread_name_prefix);
-  virtual ~TaskGraphRunner();
+void Task::DidSchedule() {
+  did_schedule_ = true;
+}
 
-  void Register(const WorkerPool* worker_pool);
-  void Unregister(const WorkerPool* worker_pool);
-  // Schedule running of tasks in |graph|. Tasks previously scheduled but
-  // no longer needed will be canceled unless already running. Canceled
-  // tasks are moved to |completed_tasks| without being run. The result
-  // is that once scheduled, a task is guaranteed to end up in the
-  // |completed_tasks| queue even if it later get canceled by another
-  // call to SetTaskGraph().
-  void SetTaskGraph(const WorkerPool* worker_pool, TaskGraph* graph);
+void Task::WillRun() {
+  DCHECK(did_schedule_);
+  DCHECK(!did_run_);
+}
 
-  // Wait for all scheduled tasks to finish running.
-  void WaitForTasksToFinishRunning(const WorkerPool* worker_pool);
+void Task::DidRun() {
+  did_run_ = true;
+}
 
-  // Collect all completed tasks in |completed_tasks|.
-  void CollectCompletedTasks(const WorkerPool* worker_pool,
-                             TaskVector* completed_tasks);
+bool Task::HasFinishedRunning() const {
+  return did_run_;
+}
 
- private:
-  static bool CompareTaskPriority(const internal::GraphNode* a,
-                                  const internal::GraphNode* b) {
-    // In this system, numerically lower priority is run first.
-    if (a->priority() != b->priority())
-      return a->priority() > b->priority();
+GraphNode::GraphNode(Task* task, unsigned priority)
+    : task_(task),
+      priority_(priority),
+      num_dependencies_(0) {
+}
 
-    // Run task with most dependents first when priority is the same.
-    return a->dependents().size() < b->dependents().size();
-  }
+GraphNode::~GraphNode() {}
 
-  struct TaskNamespace {
-    // This set contains all pending tasks.
-    TaskGraph pending_tasks;
-    // This set contains all currently running tasks.
-    TaskGraph running_tasks;
-    // Completed tasks not yet collected by origin thread.
-    TaskVector completed_tasks;
-    // Ordered set of tasks that are ready to run.
-    internal::GraphNode::Vector ready_to_run_tasks;
-  };
+TaskGraphRunner::TaskNamespace::TaskNamespace() {}
 
-  static bool CompareTaskNamespacePriority(const TaskNamespace* a,
-                                           const TaskNamespace* b) {
-    DCHECK(!a->ready_to_run_tasks.empty());
-    DCHECK(!b->ready_to_run_tasks.empty());
-
-    // Compare based on task priority of the ready_to_run_tasks heap
-    // .front() will hold the max element of the heap,
-    // except after pop_heap, when max element is moved to .back().
-    return CompareTaskPriority(a->ready_to_run_tasks.front(),
-                               b->ready_to_run_tasks.front());
-  }
-
-  typedef std::map<const WorkerPool*, linked_ptr<TaskNamespace> >
-      TaskNamespaceMap;
-
-  // Overridden from base::DelegateSimpleThread:
-  virtual void Run() OVERRIDE;
-
-  inline bool has_finished_running_tasks(TaskNamespace* task_namespace) {
-    return (task_namespace->pending_tasks.empty() &&
-            task_namespace->running_tasks.empty());
-  }
-
-  // This lock protects all members of this class except
-  // |worker_pool_on_origin_thread_|. Do not read or modify anything
-  // without holding this lock. Do not block while holding this lock.
-  mutable base::Lock lock_;
-
-  // Condition variable that is waited on by worker threads until new
-  // tasks are ready to run or shutdown starts.
-  base::ConditionVariable has_ready_to_run_tasks_cv_;
-
-  // Condition variable that is waited on by origin threads until a
-  // namespace has finished running all associated tasks.
-  base::ConditionVariable has_namespaces_with_finished_running_tasks_cv_;
-
-  // Provides each running thread loop with a unique index. First thread
-  // loop index is 0.
-  unsigned next_thread_index_;
-
-  // Set during shutdown. Tells workers to exit when no more tasks
-  // are pending.
-  bool shutdown_;
-
-  // This set contains all registered namespaces.
-  TaskNamespaceMap namespaces_;
-
-  // Ordered set of task namespaces that have ready to run tasks.
-  std::vector<TaskNamespace*> ready_to_run_namespaces_;
-
-  ScopedPtrDeque<base::DelegateSimpleThread> workers_;
-
-  DISALLOW_COPY_AND_ASSIGN(TaskGraphRunner);
-};
+TaskGraphRunner::TaskNamespace::~TaskNamespace() {}
 
 TaskGraphRunner::TaskGraphRunner(
     size_t num_threads, const std::string& thread_name_prefix)
     : lock_(),
       has_ready_to_run_tasks_cv_(&lock_),
       has_namespaces_with_finished_running_tasks_cv_(&lock_),
-      next_thread_index_(0),
+      next_namespace_id_(1),
+      next_thread_index_(0u),
       shutdown_(false) {
   base::AutoLock lock(lock_);
 
   while (workers_.size() < num_threads) {
     scoped_ptr<base::DelegateSimpleThread> worker = make_scoped_ptr(
         new base::DelegateSimpleThread(
             this,
             thread_name_prefix +
             base::StringPrintf(
                 "Worker%u",
@@ skipping 16 matching lines @@
     DCHECK(!shutdown_);
     shutdown_ = true;
 
     // Wake up a worker so it knows it should exit. This will cause all workers
     // to exit as each will wake up another worker before exiting.
     has_ready_to_run_tasks_cv_.Signal();
   }
 
   while (workers_.size()) {
     scoped_ptr<base::DelegateSimpleThread> worker = workers_.take_front();
-    // http://crbug.com/240453 - Join() is considered IO and will block this
-    // thread. See also http://crbug.com/239423 for further ideas.
+    // Join() is considered IO and will block this thread.
     base::ThreadRestrictions::ScopedAllowIO allow_io;
     worker->Join();
   }
 }
 
-void TaskGraphRunner::Register(const WorkerPool* worker_pool) {
+NamespaceToken TaskGraphRunner::GetNamespaceToken() {
   base::AutoLock lock(lock_);
 
-  DCHECK(namespaces_.find(worker_pool) == namespaces_.end());
-  linked_ptr<TaskNamespace> task_set = make_linked_ptr(new TaskNamespace());
-  namespaces_[worker_pool] = task_set;
+  NamespaceToken token(next_namespace_id_++);
+  DCHECK(namespaces_.find(token.id_) == namespaces_.end());
+  return token;
 }
 
-void TaskGraphRunner::Unregister(const WorkerPool* worker_pool) {
+void TaskGraphRunner::WaitForTasksToFinishRunning(NamespaceToken token) {
   base::AutoLock lock(lock_);
 
-  DCHECK(namespaces_.find(worker_pool) != namespaces_.end());
-  DCHECK_EQ(0u, namespaces_[worker_pool]->pending_tasks.size());
-  DCHECK_EQ(0u, namespaces_[worker_pool]->ready_to_run_tasks.size());
-  DCHECK_EQ(0u, namespaces_[worker_pool]->running_tasks.size());
-  DCHECK_EQ(0u, namespaces_[worker_pool]->completed_tasks.size());
+  DCHECK(token.IsValid());
+  TaskNamespaceMap::iterator it = namespaces_.find(token.id_);
+  if (it == namespaces_.end())
+    return;
 
-  namespaces_.erase(worker_pool);
-}
-
-void TaskGraphRunner::WaitForTasksToFinishRunning(
-    const WorkerPool* worker_pool) {
-  base::AutoLock lock(lock_);
-
-  DCHECK(namespaces_.find(worker_pool) != namespaces_.end());
-  TaskNamespace* task_namespace = namespaces_[worker_pool].get();
-
-  while (!has_finished_running_tasks(task_namespace))
+  TaskNamespace* task_namespace = it->second;
+  while (!HasFinishedRunningTasksInNamespace(task_namespace))
     has_namespaces_with_finished_running_tasks_cv_.Wait();
 
   // There may be other namespaces that have finished running
   // tasks, so wake up another origin thread.
   has_namespaces_with_finished_running_tasks_cv_.Signal();
 }
 
-void TaskGraphRunner::SetTaskGraph(const WorkerPool* worker_pool,
-                                   TaskGraph* graph) {
+void TaskGraphRunner::SetTaskGraph(NamespaceToken token, TaskGraph* graph) {
+  DCHECK(token.IsValid());
+
   TaskGraph new_pending_tasks;
   TaskGraph new_running_tasks;
 
   new_pending_tasks.swap(*graph);
 
   {
     base::AutoLock lock(lock_);
 
     DCHECK(!shutdown_);
-    DCHECK(namespaces_.find(worker_pool) != namespaces_.end());
-    TaskNamespace* task_namespace = namespaces_[worker_pool].get();
+
+    scoped_ptr<TaskNamespace> task_namespace = namespaces_.take_and_erase(
+        token.id_);
+
+    // Create task namespace if it doesn't exist.
+    if (!task_namespace)
+      task_namespace.reset(new TaskNamespace);
 
     // First remove all completed tasks from |new_pending_tasks| and
     // adjust number of dependencies.
-    for (TaskVector::iterator it = task_namespace->completed_tasks.begin();
+    for (Task::Vector::iterator it = task_namespace->completed_tasks.begin();
          it != task_namespace->completed_tasks.end(); ++it) {
-      internal::WorkerPoolTask* task = it->get();
+      Task* task = it->get();
 
-      scoped_ptr<internal::GraphNode> node = new_pending_tasks.take_and_erase(
-          task);
+      scoped_ptr<GraphNode> node = new_pending_tasks.take_and_erase(task);
       if (node) {
-        for (internal::GraphNode::Vector::const_iterator it =
-                 node->dependents().begin();
+        for (GraphNode::Vector::const_iterator it = node->dependents().begin();
              it != node->dependents().end(); ++it) {
-          internal::GraphNode* dependent_node = *it;
+          GraphNode* dependent_node = *it;
           dependent_node->remove_dependency();
         }
       }
     }
 
     // Build new running task set.
     for (TaskGraph::iterator it = task_namespace->running_tasks.begin();
          it != task_namespace->running_tasks.end(); ++it) {
-      internal::WorkerPoolTask* task = it->first;
+      Task* task = it->first;
       // Transfer scheduled task value from |new_pending_tasks| to
       // |new_running_tasks| if currently running. Value must be set to
       // NULL if |new_pending_tasks| doesn't contain task. This does
       // the right in both cases.
       new_running_tasks.set(task, new_pending_tasks.take_and_erase(task));
     }
 
     // Build new "ready to run" tasks queue.
     task_namespace->ready_to_run_tasks.clear();
     for (TaskGraph::iterator it = new_pending_tasks.begin();
          it != new_pending_tasks.end(); ++it) {
-      internal::WorkerPoolTask* task = it->first;
+      Task* task = it->first;
       DCHECK(task);
-      internal::GraphNode* node = it->second;
+      GraphNode* node = it->second;
 
       // Completed tasks should not exist in |new_pending_tasks|.
       DCHECK(!task->HasFinishedRunning());
 
       // Call DidSchedule() to indicate that this task has been scheduled.
       // Note: This is only for debugging purposes.
       task->DidSchedule();
 
       if (!node->num_dependencies())
         task_namespace->ready_to_run_tasks.push_back(node);
@@ skipping 22 matching lines @@
     // Note: old tasks are intentionally destroyed after releasing |lock_|.
     task_namespace->pending_tasks.swap(new_pending_tasks);
     task_namespace->running_tasks.swap(new_running_tasks);
 
     // If |ready_to_run_tasks| is empty, it means we either have
     // running tasks, or we have no pending tasks.
     DCHECK(!task_namespace->ready_to_run_tasks.empty() ||
            (task_namespace->pending_tasks.empty() ||
             !task_namespace->running_tasks.empty()));
 
+    // Add task namespace if not empty.
+    if (!task_namespace->pending_tasks.empty() ||
+        !task_namespace->running_tasks.empty() ||
+        !task_namespace->completed_tasks.empty()) {
+      namespaces_.set(token.id_, task_namespace.Pass());
+    }
+
     // Build new "ready to run" task namespaces queue.
     ready_to_run_namespaces_.clear();
     for (TaskNamespaceMap::iterator it = namespaces_.begin();
          it != namespaces_.end(); ++it) {
       if (!it->second->ready_to_run_tasks.empty())
-        ready_to_run_namespaces_.push_back(it->second.get());
+        ready_to_run_namespaces_.push_back(it->second);
     }
 
     // Rearrange the task namespaces in |ready_to_run_namespaces_|
     // in such a way that they form a heap.
     std::make_heap(ready_to_run_namespaces_.begin(),
                    ready_to_run_namespaces_.end(),
                    CompareTaskNamespacePriority);
 
     // If there is more work available, wake up worker thread.
     if (!ready_to_run_namespaces_.empty())
       has_ready_to_run_tasks_cv_.Signal();
   }
 }
 
 void TaskGraphRunner::CollectCompletedTasks(
-    const WorkerPool* worker_pool, TaskVector* completed_tasks) {
+    NamespaceToken token, Task::Vector* completed_tasks) {
   base::AutoLock lock(lock_);
 
+  DCHECK(token.IsValid());
+  TaskNamespaceMap::iterator it = namespaces_.find(token.id_);
+  if (it == namespaces_.end())
+    return;
+
+  TaskNamespace* task_namespace = it->second;
+
   DCHECK_EQ(0u, completed_tasks->size());
-  DCHECK(namespaces_.find(worker_pool) != namespaces_.end());
-  completed_tasks->swap(namespaces_[worker_pool]->completed_tasks);
+  completed_tasks->swap(task_namespace->completed_tasks);
+  if (!HasFinishedRunningTasksInNamespace(task_namespace))
+    return;
+
+  // Remove namespace if finished running tasks.
+  DCHECK_EQ(0u, task_namespace->pending_tasks.size());
+  DCHECK_EQ(0u, task_namespace->running_tasks.size());
+  DCHECK_EQ(0u, task_namespace->completed_tasks.size());
+  DCHECK_EQ(0u, task_namespace->ready_to_run_tasks.size());
+  namespaces_.erase(it);
 }
 
 void TaskGraphRunner::Run() {
   base::AutoLock lock(lock_);
 
   // Get a unique thread index.
   int thread_index = next_thread_index_++;
 
   while (true) {
     if (ready_to_run_namespaces_.empty()) {
@@ skipping 11 matching lines @@
                   ready_to_run_namespaces_.end(),
                   CompareTaskNamespacePriority);
     TaskNamespace* task_namespace = ready_to_run_namespaces_.back();
     ready_to_run_namespaces_.pop_back();
     DCHECK(!task_namespace->ready_to_run_tasks.empty());
 
     // Take top priority task from |ready_to_run_tasks|.
     std::pop_heap(task_namespace->ready_to_run_tasks.begin(),
                   task_namespace->ready_to_run_tasks.end(),
                   CompareTaskPriority);
-    scoped_refptr<internal::WorkerPoolTask> task(
+    scoped_refptr<Task> task(
         task_namespace->ready_to_run_tasks.back()->task());
     task_namespace->ready_to_run_tasks.pop_back();
 
     // Add task namespace back to |ready_to_run_namespaces_| if not
     // empty after taking top priority task.
     if (!task_namespace->ready_to_run_tasks.empty()) {
       ready_to_run_namespaces_.push_back(task_namespace);
       std::push_heap(ready_to_run_namespaces_.begin(),
                      ready_to_run_namespaces_.end(),
                      CompareTaskNamespacePriority);
@@ skipping 16 matching lines @@
       base::AutoUnlock unlock(lock_);
 
       task->RunOnWorkerThread(thread_index);
     }
 
     // This will mark task as finished running.
     task->DidRun();
 
     // Now iterate over all dependents to remove dependency and check
     // if they are ready to run.
-    scoped_ptr<internal::GraphNode> node =
+    scoped_ptr<GraphNode> node =
         task_namespace->running_tasks.take_and_erase(task.get());
     if (node) {
       bool ready_to_run_namespaces_has_heap_properties = true;
 
-      for (internal::GraphNode::Vector::const_iterator it =
-               node->dependents().begin();
+      for (GraphNode::Vector::const_iterator it = node->dependents().begin();
            it != node->dependents().end(); ++it) {
-        internal::GraphNode* dependent_node = *it;
+        GraphNode* dependent_node = *it;
 
         dependent_node->remove_dependency();
         // Task is ready if it has no dependencies. Add it to
         // |ready_to_run_tasks_|.
         if (!dependent_node->num_dependencies()) {
           bool was_empty = task_namespace->ready_to_run_tasks.empty();
           task_namespace->ready_to_run_tasks.push_back(dependent_node);
           std::push_heap(task_namespace->ready_to_run_tasks.begin(),
                          task_namespace->ready_to_run_tasks.end(),
                          CompareTaskPriority);
@@ skipping 17 matching lines @@
         std::make_heap(ready_to_run_namespaces_.begin(),
                        ready_to_run_namespaces_.end(),
                        CompareTaskNamespacePriority);
       }
     }
 
     // Finally add task to |completed_tasks_|.
     task_namespace->completed_tasks.push_back(task);
 
     // If namespace has finished running all tasks, wake up origin thread.
-    if (has_finished_running_tasks(task_namespace))
+    if (HasFinishedRunningTasksInNamespace(task_namespace))
       has_namespaces_with_finished_running_tasks_cv_.Signal();
   }
 
   // We noticed we should exit. Wake up the next worker so it knows it should
   // exit as well (because the Shutdown() code only signals once).
   has_ready_to_run_tasks_cv_.Signal();
 }
 
-class CompositorRasterTaskGraphRunner
-    : public TaskGraphRunner {
- public:
-  CompositorRasterTaskGraphRunner() : TaskGraphRunner(
-      WorkerPool::GetNumRasterThreads(), "CompositorRaster") {
-  }
-};
-
-base::LazyInstance<CompositorRasterTaskGraphRunner>
-    g_task_graph_runner = LAZY_INSTANCE_INITIALIZER;
-
-const int kDefaultNumRasterThreads = 1;
-
-int g_num_raster_threads = 0;
-
-}  // namespace
-
-namespace internal {
-
-WorkerPoolTask::WorkerPoolTask()
-    : did_schedule_(false),
-      did_run_(false),
-      did_complete_(false) {
-}
-
-WorkerPoolTask::~WorkerPoolTask() {
-  DCHECK_EQ(did_schedule_, did_complete_);
-  DCHECK(!did_run_ || did_schedule_);
-  DCHECK(!did_run_ || did_complete_);
-}
-
-void WorkerPoolTask::DidSchedule() {
-  DCHECK(!did_complete_);
-  did_schedule_ = true;
-}
-
-void WorkerPoolTask::WillRun() {
-  DCHECK(did_schedule_);
-  DCHECK(!did_complete_);
-  DCHECK(!did_run_);
-}
-
-void WorkerPoolTask::DidRun() {
-  did_run_ = true;
-}
-
-void WorkerPoolTask::WillComplete() {
-  DCHECK(!did_complete_);
-}
-
-void WorkerPoolTask::DidComplete() {
-  DCHECK(did_schedule_);
-  DCHECK(!did_complete_);
-  did_complete_ = true;
-}
-
-bool WorkerPoolTask::HasFinishedRunning() const {
-  return did_run_;
-}
-
-bool WorkerPoolTask::HasCompleted() const {
-  return did_complete_;
-}
-
-GraphNode::GraphNode(internal::WorkerPoolTask* task, unsigned priority)
-    : task_(task),
-      priority_(priority),
-      num_dependencies_(0) {
-}
-
-GraphNode::~GraphNode() {
-}
-
 }  // namespace internal
-
-// static
-void WorkerPool::SetNumRasterThreads(int num_threads) {
-  DCHECK_LT(0, num_threads);
-  DCHECK_EQ(0, g_num_raster_threads);
-
-  g_num_raster_threads = num_threads;
-}
-
-// static
-int WorkerPool::GetNumRasterThreads() {
-  if (!g_num_raster_threads)
-    g_num_raster_threads = kDefaultNumRasterThreads;
-
-  return g_num_raster_threads;
-}
-
-WorkerPool::WorkerPool() : in_dispatch_completion_callbacks_(false) {
-  g_task_graph_runner.Pointer()->Register(this);
-}
-
-WorkerPool::~WorkerPool() {
-  g_task_graph_runner.Pointer()->Unregister(this);
-}
-
-void WorkerPool::Shutdown() {
-  TRACE_EVENT0("cc", "WorkerPool::Shutdown");
-
-  DCHECK(!in_dispatch_completion_callbacks_);
-
-  g_task_graph_runner.Pointer()->WaitForTasksToFinishRunning(this);
-}
-
-void WorkerPool::SetTaskGraph(TaskGraph* graph) {
-  TRACE_EVENT1("cc", "WorkerPool::SetTaskGraph",
-               "num_tasks", graph->size());
-
-  DCHECK(!in_dispatch_completion_callbacks_);
-
-  g_task_graph_runner.Pointer()->SetTaskGraph(this, graph);
-}
-
-void WorkerPool::CheckForCompletedWorkerTasks() {
-  TRACE_EVENT0("cc", "WorkerPool::CheckForCompletedWorkerTasks");
-
-  DCHECK(!in_dispatch_completion_callbacks_);
-
-  TaskVector completed_tasks;
-  g_task_graph_runner.Pointer()->CollectCompletedTasks(this, &completed_tasks);
-  ProcessCompletedTasks(completed_tasks);
-}
-
-void WorkerPool::ProcessCompletedTasks(
-    const TaskVector& completed_tasks) {
-  TRACE_EVENT1("cc", "WorkerPool::ProcessCompletedTasks",
-               "completed_task_count", completed_tasks.size());
-
-  // Worker pool instance is not reentrant while processing completed tasks.
-  in_dispatch_completion_callbacks_ = true;
-
-  for (TaskVector::const_iterator it = completed_tasks.begin();
-       it != completed_tasks.end();
-       ++it) {
-    internal::WorkerPoolTask* task = it->get();
-
-    task->WillComplete();
-    task->CompleteOnOriginThread();
-    task->DidComplete();
-  }
-
-  in_dispatch_completion_callbacks_ = false;
-}
-
 }  // namespace cc