Reland: Implement and test task runners in RasterWorkerPool.

This CL moves RasterWorkerPool outside render_thread_impl.h and
in its own file, implements TaskRunner and SequencedTaskRunner
on top of it and properly tests the two implementations.

BUG=
CQ_INCLUDE_TRYBOTS=tryserver.blink:linux_blink_rel

Committed: https://crrev.com/6526c209494a5176f82b5f400fd1e66a426c840f
Cr-Commit-Position: refs/heads/master@{#341827}

Review URL: https://codereview.chromium.org/1254003003

Cr-Commit-Position: refs/heads/master@{#341946}

5 files changed, 349 insertions, 122 deletions

diff --git a/content/renderer/raster_worker_pool.cc b/content/renderer/raster_worker_pool.cc
new file mode 100644
index 0000000..978ec41
--- /dev/null
+++ b/content/renderer/raster_worker_pool.cc
@@ -0,0 +1,168 @@
+// 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 "raster_worker_pool.h"
+
+#include "base/strings/stringprintf.h"
+
+namespace content {
+
+// A sequenced task runner which posts tasks to a RasterWorkerPool.
+class RasterWorkerPool::RasterWorkerPoolSequencedTaskRunner
+    : public base::SequencedTaskRunner {
+ public:
+  RasterWorkerPoolSequencedTaskRunner(cc::TaskGraphRunner* task_graph_runner)
+      : task_graph_runner_(task_graph_runner),
+        namespace_token_(task_graph_runner->GetNamespaceToken()) {}
+
+  // Overridden from base::TaskRunner:
+  bool PostDelayedTask(const tracked_objects::Location& from_here,
+                       const base::Closure& task,
+                       base::TimeDelta delay) override {
+    return PostNonNestableDelayedTask(from_here, task, delay);
+  }
+  bool RunsTasksOnCurrentThread() const override { return true; }
+
+  // Overridden from base::SequencedTaskRunner:
+  bool PostNonNestableDelayedTask(const tracked_objects::Location& from_here,
+                                  const base::Closure& task,
+                                  base::TimeDelta delay) override {
+    base::AutoLock lock(lock_);
+
+    // Remove completed tasks.
+    DCHECK(completed_tasks_.empty());
+    task_graph_runner_->CollectCompletedTasks(namespace_token_,
+                                              &completed_tasks_);
+
+    tasks_.erase(tasks_.begin(), tasks_.begin() + completed_tasks_.size());
+
+    tasks_.push_back(make_scoped_refptr(new ClosureTask(task)));
+    graph_.Reset();
+    for (const auto& task : tasks_) {
+      int dependencies = 0;
+      if (!graph_.nodes.empty())
+        dependencies = 1;
+
+      cc::TaskGraph::Node node(task.get(), 0, dependencies);
+      if (dependencies) {
+        graph_.edges.push_back(
+            cc::TaskGraph::Edge(graph_.nodes.back().task, node.task));
+      }
+      graph_.nodes.push_back(node);
+    }
+    task_graph_runner_->ScheduleTasks(namespace_token_, &graph_);
+    completed_tasks_.clear();
+    return true;
+  }
+
+ private:
+  ~RasterWorkerPoolSequencedTaskRunner() override {
+    task_graph_runner_->WaitForTasksToFinishRunning(namespace_token_);
+    task_graph_runner_->CollectCompletedTasks(namespace_token_,
+                                              &completed_tasks_);
+  };
+
+  cc::TaskGraphRunner* const task_graph_runner_;
+
+  // Lock to exclusively access all the following members that are used to
+  // implement the SequencedTaskRunner interfaces.
+  base::Lock lock_;
+  // Namespace used to schedule tasks in the task graph runner.
+  cc::NamespaceToken namespace_token_;
+  // List of tasks currently queued up for execution.
+  cc::Task::Vector tasks_;
+  // Graph object used for scheduling tasks.
+  cc::TaskGraph graph_;
+  // Cached vector to avoid allocation when getting the list of complete
+  // tasks.
+  cc::Task::Vector completed_tasks_;
+};
+
+RasterWorkerPool::RasterWorkerPool()
+    : namespace_token_(task_graph_runner_.GetNamespaceToken()) {}
+
+void RasterWorkerPool::Start(
+    int num_threads,
+    const base::SimpleThread::Options& thread_options) {
+  DCHECK(threads_.empty());
+  while (threads_.size() < static_cast<size_t>(num_threads)) {
+    scoped_ptr<base::DelegateSimpleThread> thread(
+        new base::DelegateSimpleThread(
+            this, base::StringPrintf("CompositorTileWorker%u",
+                                     static_cast<unsigned>(threads_.size() + 1))
+                      .c_str(),
+            thread_options));
+    thread->Start();
+    threads_.push_back(thread.Pass());
+  }
+}
+
+void RasterWorkerPool::Shutdown() {
+  task_graph_runner_.WaitForTasksToFinishRunning(namespace_token_);
+  task_graph_runner_.CollectCompletedTasks(namespace_token_, &completed_tasks_);
+  // Shutdown raster threads.
+  task_graph_runner_.Shutdown();
+  while (!threads_.empty()) {
+    threads_.back()->Join();
+    threads_.pop_back();
+  }
+}
+
+// Overridden from base::TaskRunner:
+bool RasterWorkerPool::PostDelayedTask(
+    const tracked_objects::Location& from_here,
+    const base::Closure& task,
+    base::TimeDelta delay) {
+  base::AutoLock lock(lock_);
+
+  // Remove completed tasks.
+  DCHECK(completed_tasks_.empty());
+  task_graph_runner_.CollectCompletedTasks(namespace_token_, &completed_tasks_);
+
+  cc::Task::Vector::iterator end = std::remove_if(
+      tasks_.begin(), tasks_.end(), [this](const scoped_refptr<cc::Task>& e) {
+        return std::find(this->completed_tasks_.begin(),
+                         this->completed_tasks_.end(),
+                         e) != this->completed_tasks_.end();
+      });
+  tasks_.erase(end, tasks_.end());
+
+  tasks_.push_back(make_scoped_refptr(new ClosureTask(task)));
+  graph_.Reset();
+  for (const auto& task : tasks_)
+    graph_.nodes.push_back(cc::TaskGraph::Node(task.get(), 0, 0));
+
+  task_graph_runner_.ScheduleTasks(namespace_token_, &graph_);
+  completed_tasks_.clear();
+  return true;
+}
+
+bool RasterWorkerPool::RunsTasksOnCurrentThread() const {
+  return true;
+}
+
+// Overridden from base::DelegateSimpleThread::Delegate:
+void RasterWorkerPool::Run() {
+  task_graph_runner_.Run();
+}
+
+scoped_refptr<base::SequencedTaskRunner>
+RasterWorkerPool::CreateSequencedTaskRunner() {
+  return new RasterWorkerPoolSequencedTaskRunner(&task_graph_runner_);
+}
+
+RasterWorkerPool::~RasterWorkerPool() {}
+
+RasterWorkerPool::ClosureTask::ClosureTask(const base::Closure& closure)
+    : closure_(closure) {}
+
+// Overridden from cc::Task:
+void RasterWorkerPool::ClosureTask::RunOnWorkerThread() {
+  closure_.Run();
+  closure_.Reset();
+};
+
+RasterWorkerPool::ClosureTask::~ClosureTask() {}
+
+}  // namespace content
diff --git a/content/renderer/raster_worker_pool.h b/content/renderer/raster_worker_pool.h
new file mode 100644
index 0000000..b8b4f35
--- /dev/null
+++ b/content/renderer/raster_worker_pool.h
@@ -0,0 +1,106 @@
+// 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.
+
+#ifndef CONTENT_RENDERER_RASTER_WORKER_POOL_H_
+#define CONTENT_RENDERER_RASTER_WORKER_POOL_H_
+
+#include "base/callback.h"
+#include "base/containers/hash_tables.h"
+#include "base/memory/scoped_vector.h"
+#include "base/sequenced_task_runner.h"
+#include "base/task_runner.h"
+#include "base/threading/simple_thread.h"
+#include "cc/raster/task_graph_runner.h"
+#include "content/common/content_export.h"
+
+namespace content {
+
+// A pool of threads used to run raster work.
+// Work can be scheduled on the threads using different interfaces.
+// The pool itself implements TaskRunner interface and tasks posted via that
+// interface might run in parallel.
+// CreateSequencedTaskRunner creates a sequenced task runner that might run in
+// parallel with other instances of sequenced task runners.
+// It's also possible to get the underlying TaskGraphRunner to schedule a graph
+// of tasks with their dependencies.
+// TODO(reveman): make TaskGraphRunner an abstract interface and have this
+// WorkerPool class implement it.
+class CONTENT_EXPORT RasterWorkerPool
+    : public base::TaskRunner,
+      public base::DelegateSimpleThread::Delegate {
+ public:
+  RasterWorkerPool();
+
+  // Overridden from base::TaskRunner:
+  bool PostDelayedTask(const tracked_objects::Location& from_here,
+                       const base::Closure& task,
+                       base::TimeDelta delay) override;
+  bool RunsTasksOnCurrentThread() const override;
+
+  // Overridden from base::DelegateSimpleThread::Delegate:
+  void Run() override;
+
+  // Spawn |num_threads| number of threads and start running work on the
+  // worker threads.
+  void Start(int num_threads,
+             const base::SimpleThread::Options& thread_options);
+
+  // Finish running all the posted tasks (and nested task posted by those tasks)
+  // of all the associated task runners.
+  // Once all the tasks are executed the method blocks until the threads are
+  // terminated.
+  void Shutdown();
+
+  cc::TaskGraphRunner* GetTaskGraphRunner() { return &task_graph_runner_; }
+
+  // Create a new sequenced task graph runner.
+  scoped_refptr<base::SequencedTaskRunner> CreateSequencedTaskRunner();
+
+ protected:
+  ~RasterWorkerPool() override;
+
+ private:
+  class RasterWorkerPoolSequencedTaskRunner;
+  friend class RasterWorkerPoolSequencedTaskRunner;
+
+  // Simple Task for the TaskGraphRunner that wraps a closure.
+  // This class is used to schedule TaskRunner tasks on the
+  // |task_graph_runner_|.
+  class ClosureTask : public cc::Task {
+   public:
+    explicit ClosureTask(const base::Closure& closure);
+
+    // Overridden from cc::Task:
+    void RunOnWorkerThread() override;
+
+   protected:
+    ~ClosureTask() override;
+
+   private:
+    base::Closure closure_;
+
+    DISALLOW_COPY_AND_ASSIGN(ClosureTask);
+  };
+
+  // The actual threads where work is done.
+  ScopedVector<base::DelegateSimpleThread> threads_;
+  cc::TaskGraphRunner task_graph_runner_;
+
+  // Lock to exclusively access all the following members that are used to
+  // implement the TaskRunner interfaces.
+  base::Lock lock_;
+  // Namespace used to schedule tasks in the task graph runner.
+  cc::NamespaceToken namespace_token_;
+  // List of tasks currently queued up for execution.
+  cc::Task::Vector tasks_;
+  // Graph object used for scheduling tasks.
+  cc::TaskGraph graph_;
+  // Cached vector to avoid allocation when getting the list of complete
+  // tasks.
+  cc::Task::Vector completed_tasks_;
+};
+
+}  // namespace content
+
+#endif  // CONTENT_RENDERER_RASTER_WORKER_POOL_H_
diff --git a/content/renderer/raster_worker_pool_unittest.cc b/content/renderer/raster_worker_pool_unittest.cc
new file mode 100644
index 0000000..926ba2b
--- /dev/null
+++ b/content/renderer/raster_worker_pool_unittest.cc
@@ -0,0 +1,71 @@
+// 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 "base/test/sequenced_task_runner_test_template.h"
+#include "base/test/task_runner_test_template.h"
+#include "base/threading/simple_thread.h"
+#include "content/renderer/raster_worker_pool.h"
+
+namespace base {
+namespace {
+
+// Number of threads spawned in tests.
+const int kNumThreads = 4;
+
+class RasterWorkerPoolTestDelegate {
+ public:
+  RasterWorkerPoolTestDelegate()
+      : raster_worker_pool_(new content::RasterWorkerPool()) {}
+
+  void StartTaskRunner() {
+    raster_worker_pool_->Start(kNumThreads, SimpleThread::Options());
+  }
+
+  scoped_refptr<content::RasterWorkerPool> GetTaskRunner() {
+    return raster_worker_pool_;
+  }
+
+  void StopTaskRunner() {
+    raster_worker_pool_->GetTaskGraphRunner()->FlushForTesting();
+  }
+
+  ~RasterWorkerPoolTestDelegate() { raster_worker_pool_->Shutdown(); }
+
+ private:
+  scoped_refptr<content::RasterWorkerPool> raster_worker_pool_;
+};
+
+INSTANTIATE_TYPED_TEST_CASE_P(RasterWorkerPool,
+                              TaskRunnerTest,
+                              RasterWorkerPoolTestDelegate);
+
+class RasterWorkerPoolSequencedTestDelegate {
+ public:
+  RasterWorkerPoolSequencedTestDelegate()
+      : raster_worker_pool_(new content::RasterWorkerPool()) {}
+
+  void StartTaskRunner() {
+    raster_worker_pool_->Start(kNumThreads, SimpleThread::Options());
+  }
+
+  scoped_refptr<base::SequencedTaskRunner> GetTaskRunner() {
+    return raster_worker_pool_->CreateSequencedTaskRunner();
+  }
+
+  void StopTaskRunner() {
+    raster_worker_pool_->GetTaskGraphRunner()->FlushForTesting();
+  }
+
+  ~RasterWorkerPoolSequencedTestDelegate() { raster_worker_pool_->Shutdown(); }
+
+ private:
+  scoped_refptr<content::RasterWorkerPool> raster_worker_pool_;
+};
+
+INSTANTIATE_TYPED_TEST_CASE_P(RasterWorkerPool,
+                              SequencedTaskRunnerTest,
+                              RasterWorkerPoolSequencedTestDelegate);
+
+}  // namespace
+}  // namespace base
diff --git a/content/renderer/render_thread_impl.cc b/content/renderer/render_thread_impl.cc
index 0db0b29..b96b775 100644
--- a/content/renderer/render_thread_impl.cc
+++ b/content/renderer/render_thread_impl.cc
@@ -107,6 +107,7 @@
 #include "content/renderer/media/video_capture_message_filter.h"
 #include "content/renderer/net_info_helper.h"
 #include "content/renderer/p2p/socket_dispatcher.h"
+#include "content/renderer/raster_worker_pool.h"
 #include "content/renderer/render_frame_proxy.h"
 #include "content/renderer/render_process_impl.h"
 #include "content/renderer/render_view_impl.h"
@@ -422,125 +423,6 @@ void CreateEmbeddedWorkerSetup(
 
 }  // namespace
 
-class RasterWorkerPool : public base::SequencedTaskRunner,
-                         public base::DelegateSimpleThread::Delegate {
- public:
-  RasterWorkerPool()
-      : namespace_token_(task_graph_runner_.GetNamespaceToken()) {}
-
-  void Start(int num_threads,
-             const base::SimpleThread::Options& thread_options) {
-    DCHECK(threads_.empty());
-    while (threads_.size() < static_cast<size_t>(num_threads)) {
-      scoped_ptr<base::DelegateSimpleThread> thread(
-          new base::DelegateSimpleThread(
-              this, base::StringPrintf(
-                        "CompositorTileWorker%u",
-                        static_cast<unsigned>(threads_.size() + 1)).c_str(),
-              thread_options));
-      thread->Start();
-      threads_.push_back(thread.Pass());
-    }
-  }
-
-  void Shutdown() {
-    // Shutdown raster threads.
-    task_graph_runner_.Shutdown();
-    while (!threads_.empty()) {
-      threads_.back()->Join();
-      threads_.pop_back();
-    }
-  }
-
-  // Overridden from base::TaskRunner:
-  bool PostDelayedTask(const tracked_objects::Location& from_here,
-                       const base::Closure& task,
-                       base::TimeDelta delay) override {
-    return PostNonNestableDelayedTask(from_here, task, delay);
-  }
-
-  bool RunsTasksOnCurrentThread() const override { return true; }
-
-  // Overridden from base::SequencedTaskRunner:
-  bool PostNonNestableDelayedTask(const tracked_objects::Location& from_here,
-                                  const base::Closure& task,
-                                  base::TimeDelta delay) override {
-    base::AutoLock lock(lock_);
-    DCHECK(!threads_.empty());
-
-    // Remove completed tasks.
-    DCHECK(completed_tasks_.empty());
-    task_graph_runner_.CollectCompletedTasks(namespace_token_,
-                                             &completed_tasks_);
-    DCHECK_LE(completed_tasks_.size(), tasks_.size());
-    DCHECK(std::equal(completed_tasks_.begin(), completed_tasks_.end(),
-                      tasks_.begin()));
-    tasks_.erase(tasks_.begin(), tasks_.begin() + completed_tasks_.size());
-    completed_tasks_.clear();
-
-    tasks_.push_back(make_scoped_refptr(new ClosureTask(task)));
-
-    graph_.Reset();
-    for (const auto& task : tasks_) {
-      cc::TaskGraph::Node node(task.get(), 0, graph_.nodes.size());
-      if (graph_.nodes.size()) {
-        graph_.edges.push_back(
-            cc::TaskGraph::Edge(graph_.nodes.back().task, node.task));
-      }
-      graph_.nodes.push_back(node);
-    }
-
-    task_graph_runner_.ScheduleTasks(namespace_token_, &graph_);
-    return true;
-  }
-
-  // Overridden from base::DelegateSimpleThread::Delegate:
-  void Run() override { task_graph_runner_.Run(); }
-
-  cc::TaskGraphRunner* GetTaskGraphRunner() { return &task_graph_runner_; }
-
- protected:
-  ~RasterWorkerPool() override {}
-
- private:
-  // Simple Task for the TaskGraphRunner that wraps a closure.
-  class ClosureTask : public cc::Task {
-   public:
-    ClosureTask(const base::Closure& closure) : closure_(closure) {}
-
-    // Overridden from cc::Task:
-    void RunOnWorkerThread() override {
-      closure_.Run();
-      closure_.Reset();
-    };
-
-   protected:
-    ~ClosureTask() override {}
-
-   private:
-    base::Closure closure_;
-
-    DISALLOW_COPY_AND_ASSIGN(ClosureTask);
-  };
-
-  // The actual threads where work is done.
-  ScopedVector<base::DelegateSimpleThread> threads_;
-  cc::TaskGraphRunner task_graph_runner_;
-
-  // Namespace where the SequencedTaskRunner tasks run.
-  const cc::NamespaceToken namespace_token_;
-
-  // Lock to exclusively access all the following members that are used to
-  // implement the SequencedTaskRunner interface.
-  base::Lock lock_;
-  // List of tasks currently queued up for execution.
-  ClosureTask::Vector tasks_;
-  // Cached vector to avoid allocation when getting the list of complete tasks.
-  ClosureTask::Vector completed_tasks_;
-  // Graph object used for scheduling tasks.
-  cc::TaskGraph graph_;
-};
-
 // For measuring memory usage after each task. Behind a command line flag.
 class MemoryObserver : public base::MessageLoop::TaskObserver {
  public:
@@ -1954,7 +1836,7 @@ RenderThreadImpl::GetMediaThreadTaskRunner() {
   return media_thread_->task_runner();
 }
 
-base::SequencedTaskRunner* RenderThreadImpl::GetWorkerSequencedTaskRunner() {
+base::TaskRunner* RenderThreadImpl::GetWorkerTaskRunner() {
   return raster_worker_pool_.get();
 }
 
diff --git a/content/renderer/render_thread_impl.h b/content/renderer/render_thread_impl.h
index 8e99eaf..eb6516e 100644
--- a/content/renderer/render_thread_impl.h
+++ b/content/renderer/render_thread_impl.h
@@ -321,8 +321,8 @@ class CONTENT_EXPORT RenderThreadImpl
   // on the renderer's main thread.
   scoped_refptr<base::SingleThreadTaskRunner> GetMediaThreadTaskRunner();
 
-  // A SequencedTaskRunner instance that runs tasks on the raster worker pool.
-  base::SequencedTaskRunner* GetWorkerSequencedTaskRunner();
+  // A TaskRunner instance that runs tasks on the raster worker pool.
+  base::TaskRunner* GetWorkerTaskRunner();
 
   // Causes the idle handler to skip sending idle notifications
   // on the two next scheduled calls, so idle notifications are