Re-land: cc: Cancel and re-prioritize worker pool tasks.

cc/base/worker_pool.h

Index: cc/base/worker_pool.h
diff --git a/cc/base/worker_pool.h b/cc/base/worker_pool.h
index aa49ec9f9bfcf18fc208ab24dcf2201ef4c9e0c6..94c3a611e2e71d60a4ab53dc723f6e686698a115 100644
--- a/cc/base/worker_pool.h
+++ b/cc/base/worker_pool.h
@@ -5,33 +5,141 @@
 #ifndef CC_BASE_WORKER_POOL_H_
 #define CC_BASE_WORKER_POOL_H_
 
+#include <deque>
 #include <string>
 
 #include "base/cancelable_callback.h"
+#include "base/memory/ref_counted.h"
 #include "base/memory/scoped_ptr.h"
 #include "base/memory/weak_ptr.h"
 #include "base/message_loop.h"
 #include "cc/base/cc_export.h"
-#include "cc/base/scoped_ptr_deque.h"
 
 namespace cc {
 
 namespace internal {
 
-class WorkerPoolTask {
+class CC_EXPORT WorkerPoolTask
+    : public base::RefCountedThreadSafe<WorkerPoolTask> {
  public:
-  virtual ~WorkerPoolTask();
-
   virtual void RunOnThread(unsigned thread_index) = 0;
+  virtual void DispatchCompletionCallback() = 0;
 
+  void DidSchedule();
+  void WillRun();
+  void DidRun();
   void DidComplete();
 
+  bool HasFinished() const;
+
  protected:
-  explicit WorkerPoolTask(const base::Closure& reply);
+  friend class base::RefCountedThreadSafe<WorkerPoolTask>;
+
+  WorkerPoolTask();
+  virtual ~WorkerPoolTask();
 
-  const base::Closure reply_;
+ private:
+  bool did_schedule_;
+  bool did_run_;
+  bool did_complete_;
 };
 

[nduca, 2013/05/21 00:31:13]

I guess i'm still puzzled why we have dependency separate from task. Why not
have WorkerPoolTask have a m_dependencies<WorkerPoolTask*>?

[reveman, 2013/05/21 01:22:36]

There's a 1-N relationship between tasks and dependency graphs.

A graph is a description of dependencies and priorities of tasks that is used by
the worker pool to determine what task to run. To change this description, we
build a new dependency graph that in one atomic operation ("atomic" from the
workers point of view) replaces the old dependency graph. This new graph will
often reference existing tasks already present in the currently active graph.
This allows us to efficiently change dependencies and priorities without
stopping the workers (except for the quick graph swap operation).

+class CC_EXPORT WorkerPoolTaskDependency {
+ public:
+  // Depth-first traversal of dependency graph. Parent nodes are visited
+  // before children.
+  class Iterator {
+   public:
+    explicit Iterator(const WorkerPoolTaskDependency* root);
+    ~Iterator();
+
+    const WorkerPoolTaskDependency* operator->() const {
+      DCHECK(current_);
+      return current_;
+    }
+
+    const WorkerPoolTaskDependency* operator*() const {
+      DCHECK(current_);
+      return current_;
+    }
+
+    Iterator& operator++() {
+      if (current_->first_child_) {
+        current_ = current_->first_child_.get();
+        return *this;
+      }
+
+      while (!current_->next_sibling_ && current_ != root_)
+        current_ = current_->parent_;
+      if (current_ == root_)
+        return *this;
+
+      current_ = current_->next_sibling_.get();
+      return *this;
+    }
+
+    operator bool() const {
+      return current_ != root_;
+    }
+
+   private:
+    const WorkerPoolTaskDependency* current_;
+    const WorkerPoolTaskDependency* root_;
+  };
+
+  // Visits each task for all direct children.
+  class TaskIterator {
+   public:
+    explicit TaskIterator(const WorkerPoolTaskDependency* parent);
+    ~TaskIterator();
+
+    WorkerPoolTask* operator->() const {
+      DCHECK(current_);
+      return current_->task_.get();
+    }
+
+    WorkerPoolTask* operator*() const {
+      DCHECK(current_);
+      return current_->task_.get();
+    }
+
+    TaskIterator& operator++() {
+      current_ = current_->next_sibling_.get();
+      return *this;
+    }
+
+    operator bool() const {
+      return !!current_;
+    }
+
+   private:
+    const WorkerPoolTaskDependency* current_;
+  };
+
+  WorkerPoolTaskDependency();
+  ~WorkerPoolTaskDependency();
+
+  static void Create(WorkerPoolTask* task,
+                     WorkerPoolTaskDependency* parent,
+                     WorkerPoolTaskDependency* dependencies);
+
+  void Swap(WorkerPoolTaskDependency* other);
+
+  WorkerPoolTask* task() const { return task_.get(); }
+
+ protected:
+  // |parent_| is only set when "parent_->last_child_ == this".
+  // This enough to traverse graph in O(n) time and O(1) space complexity,
+  // while also allowing Swap() to be implemented in O(1) time.
+  WorkerPoolTaskDependency* parent_;
+  WorkerPoolTaskDependency* last_child_;
+  scoped_ptr<WorkerPoolTaskDependency> first_child_;
+  scoped_ptr<WorkerPoolTaskDependency> next_sibling_;
+  scoped_refptr<WorkerPoolTask> task_;
+};
+
+typedef WorkerPoolTaskDependency WorkerPoolTaskGraph;
+
 }  // namespace internal
 
 class CC_EXPORT WorkerPoolClient {
@@ -42,30 +150,15 @@ class CC_EXPORT WorkerPoolClient {
   virtual ~WorkerPoolClient() {}
 };
 
-// A worker thread pool that runs rendering tasks and guarantees completion
-// of all pending tasks at shutdown.
+// A worker thread pool that runs tasks provided by task graph and
+// guarantees completion of all pending tasks at shutdown.
 class CC_EXPORT WorkerPool {
  public:
-  typedef base::Callback<void()> Callback;
-
   virtual ~WorkerPool();
 
-  static scoped_ptr<WorkerPool> Create(
-      size_t num_threads,
-      base::TimeDelta check_for_completed_tasks_delay,
-      const std::string& thread_name_prefix) {
-    return make_scoped_ptr(new WorkerPool(num_threads,
-                                          check_for_completed_tasks_delay,
-                                          thread_name_prefix));
-  }
-
   // Tells the worker pool to shutdown and returns once all pending tasks have
   // completed.
-  void Shutdown();
-
-  // Posts |task| to worker pool. On completion, |reply|
-  // is posted to the thread that called PostTaskAndReply().
-  void PostTaskAndReply(const Callback& task, const base::Closure& reply);
+  virtual void Shutdown();
 
   // Set a new client.
   void SetClient(WorkerPoolClient* client) {
@@ -77,32 +170,27 @@ class CC_EXPORT WorkerPool {
              base::TimeDelta check_for_completed_tasks_delay,
              const std::string& thread_name_prefix);
 
-  void PostTask(scoped_ptr<internal::WorkerPoolTask> task);
+  void ScheduleTasks(internal::WorkerPoolTaskGraph* task_graph);
 
  private:
   class Inner;
   friend class Inner;
 
-  void OnTaskCompleted();
-  void OnIdle();
+  typedef std::deque<scoped_refptr<internal::WorkerPoolTask> > TaskDeque;
+
+  void OnIdle(TaskDeque* completed_tasks);
   void ScheduleCheckForCompletedTasks();
   void CheckForCompletedTasks();
-  void DispatchCompletionCallbacks();
+  void DispatchCompletionCallbacks(TaskDeque* completed_tasks);
 
   WorkerPoolClient* client_;
   scoped_refptr<base::MessageLoopProxy> origin_loop_;
-  base::WeakPtrFactory<WorkerPool> weak_ptr_factory_;
+  base::CancelableClosure check_for_completed_tasks_callback_;
   base::TimeDelta check_for_completed_tasks_delay_;
   bool check_for_completed_tasks_pending_;
 
-  // Holds all completed tasks for which we have not yet dispatched
-  // reply callbacks.
-  ScopedPtrDeque<internal::WorkerPoolTask> completed_tasks_;
-
   // Hide the gory details of the worker pool in |inner_|.
   const scoped_ptr<Inner> inner_;
-
-  DISALLOW_COPY_AND_ASSIGN(WorkerPool);
 };
 
 }  // namespace cc