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/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 {
 
 WorkerPoolTask::WorkerPoolTask()
     : did_schedule_(false),
       did_run_(false),
       did_complete_(false) {
 }
@@ skipping 49 matching lines @@
 }  // 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 Shutdown(WorkerPool* wp);

[reveman, 2013/11/21 16:36:10]

I don't think we should change Shutdown(). That's confusing. I think we should
instead add a new function:

void WaitForTasksToComplete(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
   // is that once scheduled, a task is guaranteed to end up in the
   // |completed_tasks_| queue even if they later get canceled by another
   // call to SetTaskGraph().
-  void SetTaskGraph(TaskGraph* graph);
+
+  void SetTaskGraph(TaskGraph* graph, WorkerPool* wp);

[reveman, 2013/11/21 16:36:10]

Prefer if you pass WorkerPool pointer as first parameter to all these functions
and it should have the const modifier as this class should not be allowed to
modify it. Only use it as a unique task namespace key.

 
   // Collect all completed tasks in |completed_tasks|.
-  void CollectCompletedTasks(TaskVector* completed_tasks);
+  void CollectCompletedTasks(TaskVector* completed_tasks, WorkerPool* wp);

[reveman, 2013/11/21 16:36:10]

Here and above. Please don't abbreviate the parameter name. Use "worker_pool"
instead of "wp".

 
  private:
   class PriorityComparator {
    public:
     bool operator()(const internal::GraphNode* a,
                     const internal::GraphNode* b) {
       // In this system, numerically lower priority is run first.
       if (a->priority() != b->priority())
         return a->priority() > b->priority();
 
@@ skipping 15 matching lines @@
   base::ConditionVariable has_ready_to_run_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_;
 
   // 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_;

[reveman, 2013/11/21 16:36:10]

I don't think we should move any of these out of this class. Everything that can
be accessed from multiple threads should stay in this class.

Consider making each one of these something like: std::map<const WorkerPool*,
TaskVector> completed_tasks_;

 
   ScopedPtrDeque<base::DelegateSimpleThread> workers_;
 
+  // Maintain a WorkerPool instance per Inner for Task Namespace
+  WorkerPool* wp_;

[reveman, 2013/11/21 16:36:10]

If you keep a pointer to one WorkerPool instance then you're still limiting the
usage to one instance. Multiple WorkerPool instances need to be able to use this
class at the same time from different threads.

+
   DISALLOW_COPY_AND_ASSIGN(Inner);
 };
 
+class CC_EXPORT DerivedInner : public WorkerPool::Inner {
+ public:
+  DerivedInner();
+};
+
+base::LazyInstance<DerivedInner>  g_workerpool_inner;
+
 WorkerPool::Inner::Inner(
     size_t num_threads, const std::string& thread_name_prefix)
     : lock_(),
       has_ready_to_run_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, wp_->pending_tasks_.size());
   DCHECK_EQ(0u, ready_to_run_tasks_.size());
-  DCHECK_EQ(0u, running_tasks_.size());
-  DCHECK_EQ(0u, completed_tasks_.size());
+  DCHECK_EQ(0u, wp_->running_tasks_.size());
+  DCHECK_EQ(0u, wp_->completed_tasks_.size());
 }
 
-void WorkerPool::Inner::Shutdown() {
+void WorkerPool::Inner::Shutdown(WorkerPool* wp) {
   {
     base::AutoLock lock(lock_);
 
     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) {
+void WorkerPool::Inner::SetTaskGraph(TaskGraph* graph, WorkerPool* wp) {
   // It is OK to call SetTaskGraph() after shutdown if |graph| is empty.
   DCHECK(graph->empty() || !shutdown_);
 
+  wp_ = wp;

[vivekg, 2013/11/22 04:44:50]

Wouldn't it be better to have a DCHECK(wp) to ensure we have the real pointer
indeed?

   GraphNodeMap new_pending_tasks;
   GraphNodeMap new_running_tasks;
   TaskQueue new_ready_to_run_tasks;
 
   new_pending_tasks.swap(*graph);
 
   {
     base::AutoLock lock(lock_);
 
     // 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 = wp_->completed_tasks_.begin();
+         it != wp_->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 (GraphNodeMap::iterator it = wp_->running_tasks_.begin();
+         it != wp_->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();
          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);
+      wp_->pending_tasks_.erase(task);
     }
 
-    completed_tasks_.reserve(completed_tasks_.size() + pending_tasks_.size());
+    wp_->completed_tasks_.reserve(
+        wp_->completed_tasks_.size() + wp_->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();
+    for (GraphNodeMap::const_iterator it = wp_->pending_tasks_.begin();
+         it != wp_->pending_tasks_.end();
          ++it) {
-      completed_tasks_.push_back(it->first);
+      wp_->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);
+    wp_->pending_tasks_.swap(new_pending_tasks);
+    wp_->running_tasks_.swap(new_running_tasks);
     std::swap(ready_to_run_tasks_, new_ready_to_run_tasks);
 
     // 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(!ready_to_run_tasks_.empty()
+        ||(wp_->pending_tasks_.empty() || !wp_->running_tasks_.empty()));
 
     // If there is more work available, wake up worker thread.
     if (!ready_to_run_tasks_.empty())
       has_ready_to_run_tasks_cv_.Signal();
   }
 }
 
-void WorkerPool::Inner::CollectCompletedTasks(TaskVector* completed_tasks) {
+void WorkerPool::Inner::CollectCompletedTasks(
+    TaskVector* completed_tasks, WorkerPool* wp) {
   base::AutoLock lock(lock_);
-
   DCHECK_EQ(0u, completed_tasks->size());
-  completed_tasks->swap(completed_tasks_);
+  completed_tasks->swap(wp->completed_tasks_);
 }
-
 void WorkerPool::Inner::Run() {
   base::AutoLock lock(lock_);
 
   // Get a unique thread index.
   int thread_index = next_thread_index_++;
-
+ 
   while (true) {
     if (ready_to_run_tasks_.empty()) {
       // Exit when shutdown is set and no more tasks are pending.
-      if (shutdown_ && pending_tasks_.empty())
+      if (shutdown_ && wp_->pending_tasks_.empty())
         break;
 
       // Wait for more tasks.
       has_ready_to_run_tasks_cv_.Wait();
       continue;
     }
 
     // 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();
 
     // 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()));
+    DCHECK(wp_->pending_tasks_.contains(task.get()));
+    DCHECK(!wp_->running_tasks_.contains(task.get()));
+    wp_->running_tasks_.set(
+        task.get(), wp_->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(
+    scoped_ptr<internal::GraphNode> node = wp_->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);
       }
     }
 
     // Finally add task to |completed_tasks_|.
-    completed_tasks_.push_back(task);
+    wp_->completed_tasks_.push_back(task);
   }
 
   // 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))) {
+    : in_dispatch_completion_callbacks_(false) {
 }
 
 WorkerPool::~WorkerPool() {
 }
 
 void WorkerPool::Shutdown() {
   TRACE_EVENT0("cc", "WorkerPool::Shutdown");
 
   DCHECK(!in_dispatch_completion_callbacks_);
-
-  inner_->Shutdown();
+  g_workerpool_inner.Pointer()->Shutdown(this);
 }
 
 void WorkerPool::CheckForCompletedTasks() {
   TRACE_EVENT0("cc", "WorkerPool::CheckForCompletedTasks");
 
   DCHECK(!in_dispatch_completion_callbacks_);
 
   TaskVector completed_tasks;
-  inner_->CollectCompletedTasks(&completed_tasks);
+  g_workerpool_inner.Pointer()->CollectCompletedTasks(&completed_tasks, this);
+
   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;
 
   for (TaskVector::const_iterator it = completed_tasks.begin();
        it != completed_tasks.end();
        ++it) {
     internal::WorkerPoolTask* task = it->get();
 
     task->WillComplete();
     task->CompleteOnOriginThread();
     task->DidComplete();
   }
 
   in_dispatch_completion_callbacks_ = false;
 }
 
 void WorkerPool::SetTaskGraph(TaskGraph* graph) {
   TRACE_EVENT1("cc", "WorkerPool::SetTaskGraph",
                "num_tasks", graph->size());
 
   DCHECK(!in_dispatch_completion_callbacks_);
+  
+  g_workerpool_inner.Pointer()->SetTaskGraph(graph, this);
+}
 
-  inner_->SetTaskGraph(graph);
+DerivedInner::DerivedInner()
+    : Inner((size_t)cc::switches::GetNumRasterThreads(), "CompositorRaster") {
 }
 
 }  // namespace cc