Shared Raster Worker Threads

cc/resources/worker_pool.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 <algorithm>
 #include <queue>
 
 #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"
+#include "cc/base/switches.h"
 
 namespace cc {
 
-namespace internal {
+namespace {
 
-WorkerPoolTask::WorkerPoolTask()
-    : did_schedule_(false),
-      did_run_(false),
-      did_complete_(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;
 
-WorkerPoolTask::~WorkerPoolTask() {
-  DCHECK_EQ(did_schedule_, did_complete_);
-  DCHECK(!did_run_ || did_schedule_);
-  DCHECK(!did_run_ || did_complete_);
-}
+  TaskGraphRunner(size_t num_threads, const std::string& thread_name_prefix);
+  virtual ~TaskGraphRunner();
 
-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
-
-// Internal to the worker pool. Any data or logic that needs to be
-// shared between threads lives in this class. All members are guarded
-// by |lock_|.
-class WorkerPool::Inner : public base::DelegateSimpleThread::Delegate {
- public:
-  Inner(size_t num_threads, const std::string& thread_name_prefix);
-  virtual ~Inner();
-
-  void Shutdown();
+  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
+  // 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 they later get canceled by another
+  // |completed_tasks| queue even if it later get canceled by another
   // call to SetTaskGraph().
-  void SetTaskGraph(TaskGraph* graph);
+  void SetTaskGraph(const WorkerPool* worker_pool, TaskGraph* graph);
+
+  // Wait for all scheduled tasks to finish running

[reveman, 2013/12/31 01:16:52]

nit: end line with "."

[sohanjg, 2013/12/31 06:31:25]

Done.

+  void WaitForTasksToFinishRunning(const WorkerPool* worker_pool);
 
   // Collect all completed tasks in |completed_tasks|.
-  void CollectCompletedTasks(TaskVector* completed_tasks);
+  void CollectCompletedTasks(const WorkerPool* worker_pool,
+                             TaskVector* completed_tasks);
 
  private:
-  class PriorityComparator {
+  class TaskPriorityComparator {
    public:
-    bool operator()(const internal::GraphNode* a,

[vmpstr, 2013/12/30 21:13:01]

Why non-const?

[sohanjg, 2013/12/31 06:31:25]

Had to drop the const to avoid code duplication, otherwise we had to put the
same comparison check in TaskNamespacePriorityComparator, as
a->ready_to_run_tasks.top() and b->ready_to_run_tasks.top() are non-constant.

[vmpstr, 2014/01/02 05:38:55]

I'm not sure I understand. priority_queue::top() is const, I think. That and a
non-const to const cast is implicit, so it shouldn't be a problem... (I could be
missing something though :P)

-                    const internal::GraphNode* b) {
+    bool operator()(internal::GraphNode* a,
+                    internal::GraphNode* b) {
       // In this system, numerically lower priority is run first.
       if (a->priority() != b->priority())
         return a->priority() > b->priority();
 
       // Run task with most dependents first when priority is the same.
       return a->dependents().size() < b->dependents().size();
     }
   };
 
+  typedef std::priority_queue<internal::GraphNode*,
+                              std::vector<internal::GraphNode*>,
+                              TaskPriorityComparator> TaskQueue;
+
+  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.
+    TaskQueue ready_to_run_tasks;
+  };
+
+  class TaskNamespacePriorityComparator {
+   public:
+    bool operator()(TaskNamespace* a,

[vmpstr, 2013/12/30 21:13:01]

If we define GraphNode::operator<() instead of TaskPriorityComparator, then this
can just be "return a->ready_to_run_tasks.top() < b->ready_to_run_tasks.top()"
(accounting for possibility of empty queues of course) and you don't need
task_comparators_ member then.

This would make it less explicit that operator< compares priorities of two graph
nodes, but I think that's OK? Either way is fine, I guess... This is just
slightly more complicated

[sohanjg, 2013/12/31 06:31:25]

As you pointed out, doing it with "operator <()" will not be as explicit as
doing it with TaskpriorityComparator, which is what we are actually doing,
comparing the task priority of the nodes. And here comparing the TaskNamespace
priority based on  ready_to_run_tasks nodes for each WorkerPool instance.
imo this should be fine. 

[reveman, 2014/01/02 03:50:08]

Please rebase this onto:

https://codereview.chromium.org/123113002/

and add a new CompareTaskNamespacePriority function that uses
CompareTaskPriority.

+                    TaskNamespace* b) {
+      return task_comparators_.operator()(a->ready_to_run_tasks.top(),

[vmpstr, 2013/12/30 21:13:01]

DCHECK that the queues aren't empty, please. (I know that it shouldn't be
possible, but an extra check won't hurt)

[sohanjg, 2013/12/31 06:31:25]

Should we do it without holding the lock here ?

[reveman, 2014/01/02 03:50:08]

No need to worry about the lock here. That's the callers responsibility and it
should always be held when this is used.

+                                          b->ready_to_run_tasks.top());
+    }
+    TaskPriorityComparator task_comparators_;

[vmpstr, 2013/12/30 21:13:01]

nit: task_comparator_?

[sohanjg, 2013/12/31 06:31:25]

Done.

+  };
+
+  typedef std::map<const WorkerPool*, linked_ptr<TaskNamespace> >
+      TaskNamespaceMap;
+
+  typedef std::priority_queue<TaskNamespace*,
+                              std::vector<TaskNamespace*>,
+                              TaskNamespacePriorityComparator>
+      TaskNamespaceQueue;
+
   // 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 pending tasks.
-  GraphNodeMap pending_tasks_;
+  // This set contains all registered namespaces.
+  TaskNamespaceMap namespaces_;
 
-  // Ordered set of tasks that are ready to run.
-  typedef std::priority_queue<internal::GraphNode*,
-                              std::vector<internal::GraphNode*>,
-                              PriorityComparator> TaskQueue;
-  TaskQueue ready_to_run_tasks_;
-
-  // This set contains all currently running tasks.
-  GraphNodeMap running_tasks_;
-
-  // Completed tasks not yet collected by origin thread.
-  TaskVector completed_tasks_;
+  // Ordered set of tasks namespaces that have ready to run tasks.

[reveman, 2013/12/31 01:16:52]

nit: s/set of tasks namespaces/set of task namespaces/

[sohanjg, 2013/12/31 06:31:25]

Done.

+  TaskNamespaceQueue ready_to_run_namespaces_;
 
   ScopedPtrDeque<base::DelegateSimpleThread> workers_;
 
-  DISALLOW_COPY_AND_ASSIGN(Inner);
+  DISALLOW_COPY_AND_ASSIGN(TaskGraphRunner);
 };
 
-WorkerPool::Inner::Inner(
+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),
       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",
                 static_cast<unsigned>(workers_.size() + 1)).c_str()));
     worker->Start();
 #if defined(OS_ANDROID) || defined(OS_LINUX)
     worker->SetThreadPriority(base::kThreadPriority_Background);
 #endif
     workers_.push_back(worker.Pass());
   }
 }
 
-WorkerPool::Inner::~Inner() {
-  base::AutoLock lock(lock_);
-
-  DCHECK(shutdown_);
-
-  DCHECK_EQ(0u, pending_tasks_.size());
-  DCHECK_EQ(0u, ready_to_run_tasks_.size());
-  DCHECK_EQ(0u, running_tasks_.size());
-  DCHECK_EQ(0u, completed_tasks_.size());
-}
-
-void WorkerPool::Inner::Shutdown() {
+TaskGraphRunner::~TaskGraphRunner() {
   {
     base::AutoLock lock(lock_);
 
+    DCHECK_EQ(0u, ready_to_run_namespaces_.size());
+    DCHECK_EQ(0u, namespaces_.size());
+
     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.
     base::ThreadRestrictions::ScopedAllowIO allow_io;
     worker->Join();
   }
 }
 
-void WorkerPool::Inner::SetTaskGraph(TaskGraph* graph) {
-  // It is OK to call SetTaskGraph() after shutdown if |graph| is empty.
-  DCHECK(graph->empty() || !shutdown_);
+void TaskGraphRunner::Register(const WorkerPool* worker_pool) {
+  base::AutoLock lock(lock_);
 
-  GraphNodeMap new_pending_tasks;
-  GraphNodeMap new_running_tasks;
+  DCHECK(namespaces_.find(worker_pool) == namespaces_.end());
+  linked_ptr<TaskNamespace> task_set = make_linked_ptr(new TaskNamespace());
+  namespaces_[worker_pool] = task_set;
+}
+
+void TaskGraphRunner::Unregister(const WorkerPool* worker_pool) {
+  base::AutoLock lock(lock_);
+
+  DCHECK(namespaces_.find(worker_pool) != namespaces_.end());
+
+  TaskNamespace* task_namespace = namespaces_[worker_pool].get();

[vmpstr, 2013/12/30 21:13:01]

Save the iterator from find here, so that you can erase by iterator

[reveman, 2013/12/31 01:16:52]

Alternatively, use "namespaces_[worker_pool]" directly in the DCHECKs where we
don't care about performance.

[sohanjg, 2013/12/31 06:31:25]

Done.

[sohanjg, 2013/12/31 06:31:25]

Think Vlad was referring to namespaces_.find(worker_pool).
Should we still need to access "namespaces_[worker_pool]" multiple times in
DCHECK's?

[reveman, 2014/01/02 03:50:08]

Multiple lookups in DCHECKs are fine but we shouldn't have that in release
builds. If you just remove the "task_namespace = name..." line above and change
all DCHECKs to
DCHECK_EQ(0u, namespaces_[worker_pool].*_tasks.size())
then we're good, imo.

+  DCHECK_EQ(0u, task_namespace->pending_tasks.size());
+  DCHECK_EQ(0u, task_namespace->ready_to_run_tasks.size());
+  DCHECK_EQ(0u, task_namespace->running_tasks.size());
+  DCHECK_EQ(0u, task_namespace->completed_tasks.size());
+
+  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 (true) {

[vmpstr, 2013/12/30 21:13:01]

I think this is equivalent to the following:

while (!has_finished_running_tasks(task_namespace))
  has_namespace_with_finished_running_tasks_cv_.Wait();

has_namespace_with_finished_running_tasks_cv_.Signal();

[sohanjg, 2013/12/31 06:31:25]

Done.

+    if (has_finished_running_tasks(task_namespace)) {
+      // There may be other namespaces that have finished running
+      // tasks, so wake up another origin thread.
+      has_namespaces_with_finished_running_tasks_cv_.Signal();
+      return;
+    }
+    has_namespaces_with_finished_running_tasks_cv_.Wait();
+  }
+}
+
+void TaskGraphRunner::SetTaskGraph(const WorkerPool* worker_pool,
+                                   TaskGraph* graph) {
+  TaskGraph new_pending_tasks;
+  TaskGraph new_running_tasks;
   TaskQueue new_ready_to_run_tasks;
+  TaskNamespaceQueue new_ready_to_run_namespaces;
 
   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();
+
     // First remove all completed tasks from |new_pending_tasks| and
     // adjust number of dependencies.
-    for (TaskVector::iterator it = completed_tasks_.begin();
-         it != completed_tasks_.end(); ++it) {
+    for (TaskVector::iterator it = task_namespace->completed_tasks.begin();
+         it != task_namespace->completed_tasks.end(); ++it) {
       internal::WorkerPoolTask* task = it->get();
 
       scoped_ptr<internal::GraphNode> node = new_pending_tasks.take_and_erase(
           task);
       if (node) {
         for (internal::GraphNode::Vector::const_iterator it =
                  node->dependents().begin();
              it != node->dependents().end(); ++it) {
           internal::GraphNode* dependent_node = *it;
           dependent_node->remove_dependency();
         }
       }
     }
 
     // Build new running task set.
-    for (GraphNodeMap::iterator it = running_tasks_.begin();
-         it != running_tasks_.end(); ++it) {
+    for (TaskGraph::iterator it = task_namespace->running_tasks.begin();
+         it != task_namespace->running_tasks.end(); ++it) {
       internal::WorkerPoolTask* 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.
     // TODO(reveman): Create this queue when building the task graph instead.
-    for (GraphNodeMap::iterator it = new_pending_tasks.begin();
+    for (TaskGraph::iterator it = new_pending_tasks.begin();
          it != new_pending_tasks.end(); ++it) {
       internal::WorkerPoolTask* task = it->first;
       DCHECK(task);
       internal::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())
         new_ready_to_run_tasks.push(node);
 
       // Erase the task from old pending tasks.
-      pending_tasks_.erase(task);
+      task_namespace->pending_tasks.erase(task);
     }
 
-    completed_tasks_.reserve(completed_tasks_.size() + pending_tasks_.size());
+    task_namespace->completed_tasks.reserve(
+        task_namespace->completed_tasks.size() +
+        task_namespace->pending_tasks.size());
 
-    // The items left in |pending_tasks_| need to be canceled.
-    for (GraphNodeMap::const_iterator it = pending_tasks_.begin();
-         it != pending_tasks_.end();
-         ++it) {
-      completed_tasks_.push_back(it->first);
+    // The items left in |pending_tasks| need to be canceled.
+    for (TaskGraph::const_iterator it = task_namespace->pending_tasks.begin();
+         it != task_namespace->pending_tasks.end(); ++it) {
+      task_namespace->completed_tasks.push_back(it->first);
     }
 
     // Swap task sets.
     // Note: old tasks are intentionally destroyed after releasing |lock_|.
-    pending_tasks_.swap(new_pending_tasks);
-    running_tasks_.swap(new_running_tasks);
-    std::swap(ready_to_run_tasks_, new_ready_to_run_tasks);
+    task_namespace->pending_tasks.swap(new_pending_tasks);
+    task_namespace->running_tasks.swap(new_running_tasks);
+    std::swap(task_namespace->ready_to_run_tasks, new_ready_to_run_tasks);
 
-    // If |ready_to_run_tasks_| is empty, it means we either have
+    // If |ready_to_run_tasks| is empty, it means we either have
     // running tasks, or we have no pending tasks.
-    DCHECK(!ready_to_run_tasks_.empty() ||
-           (pending_tasks_.empty() || !running_tasks_.empty()));
+    DCHECK(!task_namespace->ready_to_run_tasks.empty() ||
+           (task_namespace->pending_tasks.empty() ||
+            !task_namespace->running_tasks.empty()));
+
+    // Build new "ready to run" task namespaces queue.
+    for (TaskNamespaceMap::iterator it = namespaces_.begin();
+         it != namespaces_.end(); ++it) {
+      if (!it->second->ready_to_run_tasks.empty())
+        new_ready_to_run_namespaces.push(it->second.get());
+    }
+    std::swap(ready_to_run_namespaces_, new_ready_to_run_namespaces);
 
     // If there is more work available, wake up worker thread.
-    if (!ready_to_run_tasks_.empty())
+    if (!ready_to_run_namespaces_.empty())
       has_ready_to_run_tasks_cv_.Signal();
   }
 }
 
-void WorkerPool::Inner::CollectCompletedTasks(TaskVector* completed_tasks) {
+void TaskGraphRunner::CollectCompletedTasks(
+    const WorkerPool* worker_pool, TaskVector* completed_tasks) {
   base::AutoLock lock(lock_);
 
   DCHECK_EQ(0u, completed_tasks->size());
-  completed_tasks->swap(completed_tasks_);
+  DCHECK(namespaces_.find(worker_pool) != namespaces_.end());
+  completed_tasks->swap(namespaces_[worker_pool]->completed_tasks);
 }
 
-void WorkerPool::Inner::Run() {
+void TaskGraphRunner::Run() {
   base::AutoLock lock(lock_);
 
   // Get a unique thread index.
   int thread_index = next_thread_index_++;
 
   while (true) {
-    if (ready_to_run_tasks_.empty()) {
+    if (ready_to_run_namespaces_.empty()) {
       // Exit when shutdown is set and no more tasks are pending.
-      if (shutdown_ && pending_tasks_.empty())
+      if (shutdown_)
         break;
 
       // Wait for more tasks.
       has_ready_to_run_tasks_cv_.Wait();
       continue;
     }
 
-    // Take top priority task from |ready_to_run_tasks_|.
+    // Take top priority TaskNamespace from |ready_to_run_namespaces_|.
+    TaskNamespace* task_namespace = ready_to_run_namespaces_.top();
+    ready_to_run_namespaces_.pop();
+    DCHECK(!task_namespace->ready_to_run_tasks.empty());
+
+    // Take top priority task from |ready_to_run_tasks|.
     scoped_refptr<internal::WorkerPoolTask> task(
-        ready_to_run_tasks_.top()->task());
-    ready_to_run_tasks_.pop();
+        task_namespace->ready_to_run_tasks.top()->task());
+    task_namespace->ready_to_run_tasks.pop();
 
-    // Move task from |pending_tasks_| to |running_tasks_|.
-    DCHECK(pending_tasks_.contains(task.get()));
-    DCHECK(!running_tasks_.contains(task.get()));
-    running_tasks_.set(task.get(), pending_tasks_.take_and_erase(task.get()));
+    // 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(task_namespace);
+
+    // Move task from |pending_tasks| to |running_tasks|.
+    DCHECK(task_namespace->pending_tasks.contains(task.get()));
+    DCHECK(!task_namespace->running_tasks.contains(task.get()));
+    task_namespace->running_tasks.set(
+        task.get(),
+        task_namespace->pending_tasks.take_and_erase(task.get()));
 
     // There may be more work available, so wake up another worker thread.
     has_ready_to_run_tasks_cv_.Signal();
 
     // Call WillRun() before releasing |lock_| and running task.
     task->WillRun();
 
     {
       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 = running_tasks_.take_and_erase(
-        task.get());
+    scoped_ptr<internal::GraphNode> node =
+        task_namespace->running_tasks.take_and_erase(task.get());
     if (node) {
       for (internal::GraphNode::Vector::const_iterator it =
                node->dependents().begin();
            it != node->dependents().end(); ++it) {
         internal::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())
-          ready_to_run_tasks_.push(dependent_node);
+        // |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(dependent_node);
+          // Task namespace is ready if it has at least one ready
+          // to run task. Add it to |ready_to_run_namespaces_| if
+          // it just become ready.
+          if (was_empty)
+            ready_to_run_namespaces_.push(task_namespace);

[vmpstr, 2013/12/30 21:13:01]

I think this might not be correct. By adding a new ready_to_run_tasks node, you
might be changing the relative priority of the task_namespace (ie, the new node
is now top of the queue in the namespace, so the whole namespace has a higher
priority). So, if the ready_to_run_namespaces already contained task_namespace,
its position in the queue might not be correct.

[reveman, 2013/12/31 01:16:52]

Good call. I guess we'll have to iterate over all namespaces and build a new
ready_to_run_namespaces queue after having added one or more dependent nodes to
a namespace with a non-empty |ready_to_run_tasks|.

[reveman, 2013/12/31 03:37:29]

An alternative would be to just make ready_to_run_namespaces a std::vector and
just sort it when needed. That might just be the cleanest and most efficient.

[sohanjg, 2013/12/31 06:31:25]

I am not too sure about this.
When we push a new ready_to_run_tasks node it will also do a push_heap to
re-order the queue based on node priority, and it may not be top of the queue in
namespace.
And, if ready_to_run_namespaces already has task_namespace, it will again be
re-ordered correctly based on priority of the top node of ready_to_run_tasks.
Am i missing something here ?

[reveman, 2014/01/02 03:50:08]

The priority queue will not detect changes to elements already in the queue. If
the top task of a namespace already in ready_to_run_namespaces changes, then
ready_to_run_namespaces will be incorrect.

I think we should use std::make_heap, std::pop_heap, and std::push_heap directly
instead of std::priority_queue. That allows us to call std::make_heap when
necessary, which is more efficient than building a new std::priority_queue. This
makes sense on its own so I put together a patch that we can land before shared
worker threads:
https://codereview.chromium.org/123113002/

Please rebase this patch onto that and switch to using std::make_heap etc for
|ready_to_run_task_namespaces_| in a way that's consistent with how
|ready_to_run_tasks_| is used.

[vmpstr, 2014/01/02 05:29:16]

I think make_heap is a good solution to this. I don't think we expect too many
namespaces for this to be inefficient, right? I mean we could even sort every
time, but I think make_heap is better.

+        }
       }
     }
 
-    // Finally add task to |completed_tasks_|.
-    completed_tasks_.push_back(task);
+    // 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))
+      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();
 }
 
-WorkerPool::WorkerPool(size_t num_threads,
-                       const std::string& thread_name_prefix)
-    : in_dispatch_completion_callbacks_(false),
-      inner_(make_scoped_ptr(new Inner(num_threads, thread_name_prefix))) {
+class CC_EXPORT CompositorRasterTaskGraphRunner
+    : public TaskGraphRunner {
+ public:
+  CompositorRasterTaskGraphRunner() : TaskGraphRunner(
+      switches::GetNumRasterThreads(), "CompositorRaster") {

[reveman, 2014/01/02 03:50:08]

FYI, we're probably going to have to remove the use of switches:: here and
instead add some API for setting number of raster threads and retrieving the
value here and in PicturePile::Update().

What has to be done depends on the discussion here:
https://codereview.chromium.org/122063002/

+  }
+};
+
+base::LazyInstance<CompositorRasterTaskGraphRunner>
+    g_task_graph_runner = LAZY_INSTANCE_INITIALIZER;
+
+}  // 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
+
+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_);
 
-  inner_->Shutdown();
+  g_task_graph_runner.Pointer()->WaitForTasksToFinishRunning(this);
 }
 
 void WorkerPool::CheckForCompletedTasks() {
   TRACE_EVENT0("cc", "WorkerPool::CheckForCompletedTasks");
 
   DCHECK(!in_dispatch_completion_callbacks_);
 
   TaskVector completed_tasks;
-  inner_->CollectCompletedTasks(&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;
@@ skipping 10 matching lines @@
 
   in_dispatch_completion_callbacks_ = false;
 }
 
 void WorkerPool::SetTaskGraph(TaskGraph* graph) {
   TRACE_EVENT1("cc", "WorkerPool::SetTaskGraph",
                "num_tasks", graph->size());
 
   DCHECK(!in_dispatch_completion_callbacks_);
 
-  inner_->SetTaskGraph(graph);
+  g_task_graph_runner.Pointer()->SetTaskGraph(this, graph);
 }
 
 }  // namespace cc