cc: Run raster task completion callbacks during worker pool shutdown.

Re-factor RasterWorkerPool such that we can properly run all task
completion callbacks synchronously. Raster task completion tasks
are posted to the compositor threads message loop. During shutdown
all such completion tasks need to be cancelled and completion
callbacks need to be run before RasterWorkerPool::Stop() returns.

BUG=173001


Review URL: https://chromiumcodereview.appspot.com/12091055

git-svn-id: svn://svn.chromium.org/chrome/trunk/src@179849 0039d316-1c4b-4281-b951-d872f2087c98

1 files changed, 213 insertions, 0 deletions

diff --git a/cc/worker_pool.cc b/cc/worker_pool.cc
new file mode 100644
index 0000000..684e15e
--- /dev/null
+++ b/cc/worker_pool.cc
@@ -0,0 +1,213 @@
+// 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/worker_pool.h"
+
+#include <algorithm>
+
+#include "base/bind.h"
+#include "base/stl_util.h"
+#include "base/stringprintf.h"
+
+#if defined(OS_ANDROID)
+// TODO(epenner): Move thread priorities to base. (crbug.com/170549)
+#include <sys/resource.h>
+#endif
+
+namespace cc {
+
+namespace {
+
+class WorkerPoolTaskImpl : public internal::WorkerPoolTask {
+ public:
+  WorkerPoolTaskImpl(const WorkerPool::Callback& task,
+                     const base::Closure& reply)
+      : internal::WorkerPoolTask(reply),
+        task_(task) {}
+
+  virtual void Run() OVERRIDE {
+    task_.Run(&rendering_stats_);
+  }
+
+ private:
+  WorkerPool::Callback task_;
+};
+
+const char* kWorkerThreadNamePrefix = "Compositor";
+
+// Allow two pending tasks per worker. This keeps resource usage
+// low while making sure workers aren't unnecessarily idle.
+const int kNumPendingTasksPerWorker = 2;
+
+}  // namespace
+
+namespace internal {
+
+WorkerPoolTask::WorkerPoolTask(const base::Closure& reply)
+    : reply_(reply) {
+}
+
+WorkerPoolTask::~WorkerPoolTask() {
+}
+
+void WorkerPoolTask::Completed() {
+  reply_.Run();
+}
+
+}  // namespace internal
+
+WorkerPool::Worker::Worker(WorkerPool* worker_pool, const std::string name)
+    : base::Thread(name.c_str()),
+      worker_pool_(worker_pool),
+      weak_ptr_factory_(ALLOW_THIS_IN_INITIALIZER_LIST(this)) {
+  Start();
+  DCHECK(IsRunning());
+}
+
+WorkerPool::Worker::~Worker() {
+  DCHECK(!IsRunning());
+  DCHECK_EQ(pending_tasks_.size(), 0);
+}
+
+void WorkerPool::Worker::StopAfterCompletingAllPendingTasks() {
+  // Signals the thread to exit and returns once all pending tasks have run.
+  Stop();
+
+  // Complete all pending tasks. The Stop() call above guarantees that
+  // all tasks have finished running.
+  while (!pending_tasks_.empty())
+    OnTaskCompleted();
+
+  // Cancel all pending replies.
+  weak_ptr_factory_.InvalidateWeakPtrs();
+}
+
+void WorkerPool::Worker::PostTask(scoped_ptr<internal::WorkerPoolTask> task) {
+  DCHECK_LT(num_pending_tasks(), kNumPendingTasksPerWorker);
+
+  message_loop_proxy()->PostTaskAndReply(
+      FROM_HERE,
+      base::Bind(&Worker::RunTask, base::Unretained(task.get())),
+      base::Bind(&Worker::OnTaskCompleted, weak_ptr_factory_.GetWeakPtr()));
+
+  pending_tasks_.push_back(task.Pass());
+
+  worker_pool_->DidNumPendingTasksChange();
+}
+
+void WorkerPool::Worker::Init() {
+#if defined(OS_ANDROID)
+  // TODO(epenner): Move thread priorities to base. (crbug.com/170549)
+  int nice_value = 10; // Idle priority.
+  setpriority(PRIO_PROCESS, base::PlatformThread::CurrentId(), nice_value);
+#endif
+}
+
+// static
+void WorkerPool::Worker::RunTask(internal::WorkerPoolTask* task) {
+  task->Run();
+}
+
+void WorkerPool::Worker::OnTaskCompleted() {
+  CHECK(!pending_tasks_.empty());
+
+  scoped_ptr<internal::WorkerPoolTask> task = pending_tasks_.take_front();
+
+  task->Completed();
+
+  rendering_stats_.totalRasterizeTime +=
+      task->rendering_stats().totalRasterizeTime;
+  rendering_stats_.totalPixelsRasterized +=
+      task->rendering_stats().totalPixelsRasterized;
+  rendering_stats_.totalDeferredImageDecodeTime +=
+      task->rendering_stats().totalDeferredImageDecodeTime;
+  rendering_stats_.totalDeferredImageDecodeCount +=
+      task->rendering_stats().totalDeferredImageDecodeCount;
+
+  worker_pool_->DidNumPendingTasksChange();
+}
+
+WorkerPool::WorkerPool(size_t num_threads)
+    : workers_need_sorting_(false),
+      shutdown_(false) {
+  const std::string thread_name_prefix = kWorkerThreadNamePrefix;
+  while (workers_.size() < num_threads) {
+    int thread_number = workers_.size() + 1;
+    workers_.push_back(
+        new Worker(this,
+                   thread_name_prefix +
+                   StringPrintf("Worker%d", thread_number).c_str()));
+  }
+}
+
+WorkerPool::~WorkerPool() {
+  Shutdown();
+  STLDeleteElements(&workers_);
+}
+
+void WorkerPool::Shutdown() {
+  DCHECK(!shutdown_);
+  shutdown_ = true;
+
+  for (WorkerVector::iterator it = workers_.begin();
+       it != workers_.end(); it++) {
+    Worker* worker = *it;
+    worker->StopAfterCompletingAllPendingTasks();
+  }
+}
+
+void WorkerPool::PostTaskAndReply(
+    const Callback& task, const base::Closure& reply) {
+  Worker* worker = GetWorkerForNextTask();
+
+  worker->PostTask(
+      make_scoped_ptr(new WorkerPoolTaskImpl(
+                          task,
+                          reply)).PassAs<internal::WorkerPoolTask>());
+}
+
+bool WorkerPool::IsBusy() {
+  Worker* worker = GetWorkerForNextTask();
+
+  return worker->num_pending_tasks() >= kNumPendingTasksPerWorker;
+}
+
+void WorkerPool::GetRenderingStats(RenderingStats* stats) {
+  stats->totalRasterizeTime = base::TimeDelta();
+  stats->totalPixelsRasterized = 0;
+  stats->totalDeferredImageDecodeCount = 0;
+  stats->totalDeferredImageDecodeTime = base::TimeDelta();
+  for (WorkerVector::iterator it = workers_.begin();
+       it != workers_.end(); ++it) {
+    Worker* worker = *it;
+    stats->totalRasterizeTime +=
+        worker->rendering_stats().totalRasterizeTime;
+    stats->totalPixelsRasterized +=
+        worker->rendering_stats().totalPixelsRasterized;
+    stats->totalDeferredImageDecodeCount +=
+        worker->rendering_stats().totalDeferredImageDecodeCount;
+    stats->totalDeferredImageDecodeTime +=
+        worker->rendering_stats().totalDeferredImageDecodeTime;
+  }
+}
+
+WorkerPool::Worker* WorkerPool::GetWorkerForNextTask() {
+  CHECK(!shutdown_);
+  SortWorkersIfNeeded();
+  return workers_.front();
+}
+
+void WorkerPool::DidNumPendingTasksChange() {
+  workers_need_sorting_ = true;
+}
+
+void WorkerPool::SortWorkersIfNeeded() {
+  if (!workers_need_sorting_)
+    return;
+
+  std::sort(workers_.begin(), workers_.end(), NumPendingTasksComparator());
+  workers_need_sorting_ = false;
+}
+
+}  // namespace cc