Reland - Refactor signaling in RWP

content/renderer/raster_worker_pool.cc

 // 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 "content/renderer/raster_worker_pool.h"
 
 #include <string>
 #include <utility>
 #include <vector>
 
 #include "base/strings/stringprintf.h"
 #include "base/threading/thread_restrictions.h"
 #include "base/trace_event/trace_event.h"
 #include "cc/base/math_util.h"
 #include "cc/raster/task_category.h"
 
 namespace content {
 namespace {
 
 // A thread which forwards to RasterWorkerPool::Run with the runnable
 // categories.
 class RasterWorkerPoolThread : public base::SimpleThread {
  public:
   explicit RasterWorkerPoolThread(const std::string& name_prefix,
                                   const Options& options,
                                   RasterWorkerPool* pool,
-                                  std::vector<cc::TaskCategory> categories)
-      : SimpleThread(name_prefix, options),
-        pool_(pool),
-        categories_(categories) {}
+                                  cc::TaskCategory category)
+      : SimpleThread(name_prefix, options), pool_(pool), category_(category) {}
 
-  void Run() override { pool_->Run(categories_); }
+  void Run() override { pool_->Run(category_); }
 
  private:
   RasterWorkerPool* const pool_;
-  const std::vector<cc::TaskCategory> categories_;
+  const cc::TaskCategory category_;
 };
 
 }  // namespace
 
 // A sequenced task runner which posts tasks to a RasterWorkerPool.
 class RasterWorkerPool::RasterWorkerPoolSequencedTaskRunner
     : public base::SequencedTaskRunner {
  public:
   explicit RasterWorkerPoolSequencedTaskRunner(
       cc::TaskGraphRunner* task_graph_runner)
@@ skipping 62 matching lines @@
   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_(GetNamespaceToken()),
-      has_ready_to_run_tasks_cv_(&lock_),
+      nonconcurrent_foreground_has_ready_to_run_tasks_cv_(&lock_),
+      foreground_has_ready_to_run_tasks_cv_(&lock_),
+      background_has_ready_to_run_tasks_cv_(&lock_),
+      nonconcurrent_foreground_running_task_count_(0),
+      foreground_running_task_count_(0),
+      background_running_task_count_(0),
+      target_running_task_count_(0),
       has_namespaces_with_finished_running_tasks_cv_(&lock_),
       shutdown_(false) {}
 
 void RasterWorkerPool::Start(
     int num_threads,
     const base::SimpleThread::Options& thread_options) {
   DCHECK(threads_.empty());
-  while (threads_.size() < static_cast<size_t>(num_threads)) {
-    // Determine the categories that each thread can run.
-    std::vector<cc::TaskCategory> task_categories;
 
-    // The first thread can run nonconcurrent tasks.
-    if (threads_.size() == 0) {
-      task_categories.push_back(cc::TASK_CATEGORY_NONCONCURRENT_FOREGROUND);
-    }
+  target_running_task_count_ = num_threads;
 
-    // All threads can run foreground tasks.
-    task_categories.push_back(cc::TASK_CATEGORY_FOREGROUND);
-
-    // The last thread can run background tasks.
-    if (threads_.size() == (static_cast<size_t>(num_threads) - 1)) {
-      task_categories.push_back(cc::TASK_CATEGORY_BACKGROUND);
-    }
-
+  // Start a single thread for nonconcurrent foreground work.
+  {
     scoped_ptr<base::SimpleThread> thread(new RasterWorkerPoolThread(
         base::StringPrintf("CompositorTileWorker%u",
                            static_cast<unsigned>(threads_.size() + 1))
             .c_str(),
-        thread_options, this, task_categories));
+        thread_options, this, cc::TASK_CATEGORY_NONCONCURRENT_FOREGROUND));
     thread->Start();
     threads_.push_back(std::move(thread));
   }
+
+  // start |num_threads| threads for foreground work.
+  for (int i = 0; i < num_threads; i++) {
+    scoped_ptr<base::SimpleThread> thread(new RasterWorkerPoolThread(
+        base::StringPrintf("CompositorTileWorker%u",
+                           static_cast<unsigned>(threads_.size() + 1))
+            .c_str(),
+        thread_options, this, cc::TASK_CATEGORY_FOREGROUND));
+    thread->Start();
+    threads_.push_back(std::move(thread));
+  }
+
+  // Start a single thread for background work.
+  {
+    scoped_ptr<base::SimpleThread> thread(new RasterWorkerPoolThread(
+        base::StringPrintf("CompositorTileWorker%u",
+                           static_cast<unsigned>(threads_.size() + 1))
+            .c_str(),
+        thread_options, this, cc::TASK_CATEGORY_BACKGROUND));
+    thread->Start();
+    threads_.push_back(std::move(thread));
+  }
 }
 
 void RasterWorkerPool::Shutdown() {
   WaitForTasksToFinishRunning(namespace_token_);
   CollectCompletedTasks(namespace_token_, &completed_tasks_);
   // Shutdown raster threads.
   {
     base::AutoLock lock(lock_);
 
     DCHECK(!work_queue_.HasReadyToRunTasks());
     DCHECK(!work_queue_.HasAnyNamespaces());
 
     DCHECK(!shutdown_);
     shutdown_ = true;
 
     // Wake up all workers so they exit.
-    has_ready_to_run_tasks_cv_.Broadcast();
+    nonconcurrent_foreground_has_ready_to_run_tasks_cv_.Broadcast();
+    foreground_has_ready_to_run_tasks_cv_.Broadcast();
+    background_has_ready_to_run_tasks_cv_.Broadcast();
   }
   while (!threads_.empty()) {
     threads_.back()->Join();
     threads_.pop_back();
   }
 }
 
 // Overridden from base::TaskRunner:
 bool RasterWorkerPool::PostDelayedTask(
     const tracked_objects::Location& from_here,
@@ skipping 25 matching lines @@
 
   ScheduleTasksWithLockAcquired(namespace_token_, &graph_);
   completed_tasks_.clear();
   return true;
 }
 
 bool RasterWorkerPool::RunsTasksOnCurrentThread() const {
   return true;
 }
 
-void RasterWorkerPool::Run(const std::vector<cc::TaskCategory>& categories) {
+void RasterWorkerPool::Run(cc::TaskCategory category) {
   base::AutoLock lock(lock_);
+  base::ConditionVariable& cv = ConditionVariableForCategory(category);
 
   while (true) {
-    if (!RunTaskWithLockAcquired(categories)) {
+    if (!ShouldRunTaskForCategory(category)) {
+      // We are no longer running tasks, which may allow another category to
+      // start running. Signal other worker threads.
+      SignalTaskReadyConditionVariables();
+
       // Exit when shutdown is set and no more tasks are pending.
       if (shutdown_)
         break;
 
       // Wait for more tasks.
-      has_ready_to_run_tasks_cv_.Wait();
+      cv.Wait();
       continue;
     }
+    RunTaskWithLockAcquired(category);
   }
 }
 
 void RasterWorkerPool::FlushForTesting() {
   base::AutoLock lock(lock_);
 
   while (!work_queue_.HasFinishedRunningTasksInAllNamespaces()) {
     has_namespaces_with_finished_running_tasks_cv_.Wait();
   }
 }
@@ skipping 22 matching lines @@
 }
 
 void RasterWorkerPool::ScheduleTasksWithLockAcquired(cc::NamespaceToken token,
                                                      cc::TaskGraph* graph) {
   DCHECK(token.IsValid());
   DCHECK(!cc::TaskGraphWorkQueue::DependencyMismatch(graph));
   DCHECK(!shutdown_);
 
   work_queue_.ScheduleTasks(token, graph);
 
-  // If there is more work available, wake up the other worker threads.
-  if (work_queue_.HasReadyToRunTasks())
-    has_ready_to_run_tasks_cv_.Broadcast();
+  // There may be more work available, so wake up another worker thread.
+  SignalTaskReadyConditionVariables();
 }
 
 void RasterWorkerPool::WaitForTasksToFinishRunning(cc::NamespaceToken token) {
   TRACE_EVENT0("disabled-by-default-cc.debug",
                "RasterWorkerPool::WaitForTasksToFinishRunning");
 
   DCHECK(token.IsValid());
 
   {
     base::AutoLock lock(lock_);
     base::ThreadRestrictions::ScopedAllowWait allow_wait;
 
     auto* task_namespace = work_queue_.GetNamespaceForToken(token);
 
     if (!task_namespace)
       return;
 
     while (!work_queue_.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 RasterWorkerPool::CollectCompletedTasks(
     cc::NamespaceToken token,
     cc::Task::Vector* completed_tasks) {
   TRACE_EVENT0("disabled-by-default-cc.debug",
                "RasterWorkerPool::CollectCompletedTasks");
 
   {
     base::AutoLock lock(lock_);
     CollectCompletedTasksWithLockAcquired(token, completed_tasks);
   }
 }
 
 void RasterWorkerPool::CollectCompletedTasksWithLockAcquired(
     cc::NamespaceToken token,
     cc::Task::Vector* completed_tasks) {
   DCHECK(token.IsValid());
   work_queue_.CollectCompletedTasks(token, completed_tasks);
 }
 
-bool RasterWorkerPool::RunTaskWithLockAcquired(
-    const std::vector<cc::TaskCategory>& categories) {
-  for (const auto& category : categories) {
-    if (work_queue_.HasReadyToRunTasksForCategory(category)) {
-      RunTaskInCategoryWithLockAcquired(category);
-      return true;
-    }
-  }
-  return false;
-}
-
-void RasterWorkerPool::RunTaskInCategoryWithLockAcquired(
-    cc::TaskCategory category) {
+void RasterWorkerPool::RunTaskWithLockAcquired(cc::TaskCategory category) {
   TRACE_EVENT0("toplevel", "TaskGraphRunner::RunTask");
 
   lock_.AssertAcquired();
 
   auto prioritized_task = work_queue_.GetNextTaskToRun(category);
   cc::Task* task = prioritized_task.task;
 
+  // Create a ScopedTaskCount to increment the task count while running.
+  ScopedTaskCount scoped_task_count = ScopedTaskCountForCategory(category);
+
+  // There may be more work available, so wake up another worker thread.
+  SignalTaskReadyConditionVariables();
+
   // Call WillRun() before releasing |lock_| and running task.
   task->WillRun();
 
   {
     base::AutoUnlock unlock(lock_);
 
     task->RunOnWorkerThread();
   }
 
   // This will mark task as finished running.
   task->DidRun();
 
   work_queue_.CompleteTask(prioritized_task);
 
-  // We may have just dequeued more tasks, wake up the other worker threads.
-  if (work_queue_.HasReadyToRunTasks())
-    has_ready_to_run_tasks_cv_.Broadcast();
-
   // If namespace has finished running all tasks, wake up origin threads.
   if (work_queue_.HasFinishedRunningTasksInNamespace(
           prioritized_task.task_namespace))
-    has_namespaces_with_finished_running_tasks_cv_.Broadcast();
+    has_namespaces_with_finished_running_tasks_cv_.Signal();
+}
+
+bool RasterWorkerPool::ShouldRunTaskForCategory(cc::TaskCategory category) {
+  // During shutdown we want to flush threads as fast as possible - don't bother
+  // enforcing running task limits.
+  if (!shutdown_) {
+    // Enforce running task limits.
+    int foreground_thread_count = nonconcurrent_foreground_running_task_count_ +
+                                  foreground_running_task_count_;
+    int total_thread_count =
+        foreground_thread_count + background_running_task_count_;
+    if (total_thread_count >= target_running_task_count_) {
+      return false;
+    }
+
+    if (category == cc::TASK_CATEGORY_BACKGROUND &&
+        foreground_thread_count > 0) {
+      return false;
+    }
+  }
+
+  return work_queue_.HasReadyToRunTasksForCategory(category);
+}
+
+void RasterWorkerPool::SignalTaskReadyConditionVariables() {
+  lock_.AssertAcquired();
+  if (ShouldRunTaskForCategory(cc::TASK_CATEGORY_NONCONCURRENT_FOREGROUND)) {
+    nonconcurrent_foreground_has_ready_to_run_tasks_cv_.Signal();
+  }
+  if (ShouldRunTaskForCategory(cc::TASK_CATEGORY_FOREGROUND)) {
+    foreground_has_ready_to_run_tasks_cv_.Signal();
+  }
+  if (ShouldRunTaskForCategory(cc::TASK_CATEGORY_BACKGROUND)) {
+    background_has_ready_to_run_tasks_cv_.Signal();
+  }
+}
+
+RasterWorkerPool::ScopedTaskCount RasterWorkerPool::ScopedTaskCountForCategory(
+    cc::TaskCategory category) {
+  switch (category) {
+    case cc::TASK_CATEGORY_NONCONCURRENT_FOREGROUND:
+      return ScopedTaskCount(&nonconcurrent_foreground_running_task_count_);
+    case cc::TASK_CATEGORY_FOREGROUND:
+      return ScopedTaskCount(&foreground_running_task_count_);
+    case cc::TASK_CATEGORY_BACKGROUND:
+      return ScopedTaskCount(&background_running_task_count_);
+  }
+}
+
+base::ConditionVariable& RasterWorkerPool::ConditionVariableForCategory(
+    cc::TaskCategory category) {
+  switch (category) {
+    case cc::TASK_CATEGORY_NONCONCURRENT_FOREGROUND:
+      return nonconcurrent_foreground_has_ready_to_run_tasks_cv_;
+    case cc::TASK_CATEGORY_FOREGROUND:
+      return foreground_has_ready_to_run_tasks_cv_;
+    case cc::TASK_CATEGORY_BACKGROUND:
+      return background_has_ready_to_run_tasks_cv_;
+  }
 }
 
 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