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

cc/base/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/base/worker_pool.h"
 
 #if defined(OS_ANDROID)
 // TODO(epenner): Move thread priorities to base. (crbug.com/170549)
 #include <sys/resource.h>
 #endif
 
-#include <algorithm>
+#include <set>
 
 #include "base/bind.h"
 #include "base/debug/trace_event.h"
 #include "base/stringprintf.h"
 #include "base/synchronization/condition_variable.h"
 #include "base/threading/simple_thread.h"
+#include "cc/base/scoped_ptr_deque.h"
 
 namespace cc {
 
 namespace {
 
-class WorkerPoolTaskImpl : public internal::WorkerPoolTask {
+class WorkerPoolTaskGraphImpl : public internal::WorkerPoolTaskGraph {
  public:
-  WorkerPoolTaskImpl(const WorkerPool::Callback& task,
-                     const base::Closure& reply)
-      : internal::WorkerPoolTask(reply),
-        task_(task) {}
+  WorkerPoolTaskGraphImpl() {}
 
-  virtual void RunOnThread(unsigned thread_index) OVERRIDE {
-    task_.Run();
+  // Overridden from internal::WorkerPoolTaskGraph:
+  virtual bool HasMoreTasks() OVERRIDE { return false; }
+  virtual bool HasTask(internal::WorkerPoolTask* task) OVERRIDE {
+    return false;
+  }
+  virtual internal::WorkerPoolTask* TopTask() OVERRIDE {
+    NOTREACHED();
+    return NULL;
+  }
+  virtual scoped_refptr<internal::WorkerPoolTask> TakeTask(
+      internal::WorkerPoolTask* task) OVERRIDE {
+    NOTREACHED();
+    return NULL;
   }
 
  private:
-  WorkerPool::Callback task_;
+  virtual ~WorkerPoolTaskGraphImpl() {}
 };
 
 }  // namespace
 
-namespace internal {
-
-WorkerPoolTask::WorkerPoolTask(const base::Closure& reply) : reply_(reply) {
-}
-
-WorkerPoolTask::~WorkerPoolTask() {
-}
-
-void WorkerPoolTask::DidComplete() {
-  reply_.Run();
-}
-
-}  // 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(WorkerPool* worker_pool,
         size_t num_threads,
         const std::string& thread_name_prefix);
   virtual ~Inner();
 
   void Shutdown();
 
-  void PostTask(scoped_ptr<internal::WorkerPoolTask> task);
+  // Schedule running of tasks in |task_graph|. All tasks previously
+  // scheduled but not present in |task_graph| 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 ScheduleTasks().
+  void ScheduleTasks(scoped_ptr<internal::WorkerPoolTaskGraph> task_graph);
 
-  // Appends all completed tasks to worker pool's completed tasks queue
-  // and returns true if idle.
-  bool CollectCompletedTasks();
+  // Collect all completed tasks in |completed_tasks|. Returns true if idle.
+  bool CollectCompletedTasks(TaskDeque* completed_tasks);
 
  private:
-  // Appends all completed tasks to |completed_tasks|. Lock must
-  // already be acquired before calling this function.
-  bool AppendCompletedTasksWithLockAcquired(
-      ScopedPtrDeque<internal::WorkerPoolTask>* completed_tasks);
+  // Collect all completed tasks by swapping the contents of
+  // |completed_tasks| and |completed_tasks_|. Lock must be acquired
+  // before calling this function. Returns true if idle.
+  bool CollectCompletedTasksWithLockAcquired(TaskDeque* completed_tasks);
 
   // Schedule an OnIdleOnOriginThread callback if not already pending.
   // Lock must already be acquired before calling this function.
   void ScheduleOnIdleWithLockAcquired();
   void OnIdleOnOriginThread();
 
   // Overridden from base::DelegateSimpleThread:
   virtual void Run() OVERRIDE;
 
   // Pointer to worker pool. Can only be used on origin thread.
@@ skipping 15 matching lines @@
   base::WeakPtrFactory<Inner> weak_ptr_factory_;
 
   const base::Closure on_idle_callback_;
   // Set when a OnIdleOnOriginThread() callback is pending.
   bool on_idle_pending_;
 
   // Provides each running thread loop with a unique index. First thread
   // loop index is 0.
   unsigned next_thread_index_;
 
-  // Number of tasks currently running.
-  unsigned running_task_count_;
-
   // Set during shutdown. Tells workers to exit when no more tasks
   // are pending.
   bool shutdown_;
 
-  typedef ScopedPtrDeque<internal::WorkerPoolTask> TaskDeque;
-  TaskDeque pending_tasks_;
+  // The task graph. Provides tasks in order of priority.
+  scoped_ptr<internal::WorkerPoolTaskGraph> task_graph_;
+
+  // This set contains all currently running tasks.
+  typedef std::set<internal::WorkerPoolTask*> TaskSet;
+  TaskSet running_tasks_;
+
+  // Completed tasks not yet collected by origin thread.
   TaskDeque completed_tasks_;
 
   ScopedPtrDeque<base::DelegateSimpleThread> workers_;
 
   DISALLOW_COPY_AND_ASSIGN(Inner);
 };
 
 WorkerPool::Inner::Inner(WorkerPool* worker_pool,
                          size_t num_threads,
                          const std::string& thread_name_prefix)
     : worker_pool_on_origin_thread_(worker_pool),
       lock_(),
       has_pending_tasks_cv_(&lock_),
       origin_loop_(base::MessageLoopProxy::current()),
       weak_ptr_factory_(this),
       on_idle_callback_(base::Bind(&WorkerPool::Inner::OnIdleOnOriginThread,
                                    weak_ptr_factory_.GetWeakPtr())),
       on_idle_pending_(false),
       next_thread_index_(0),
-      running_task_count_(0),
-      shutdown_(false) {
+      shutdown_(false),
+      task_graph_(new WorkerPoolTaskGraphImpl) {
   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();
     workers_.push_back(worker.Pass());
   }
 }
 
 WorkerPool::Inner::~Inner() {
   base::AutoLock lock(lock_);
 
   DCHECK(shutdown_);
 
-  // Cancel all pending callbacks.
-  weak_ptr_factory_.InvalidateWeakPtrs();
-
-  DCHECK_EQ(0u, pending_tasks_.size());
+  DCHECK(!task_graph_->HasMoreTasks());
+  DCHECK_EQ(0u, running_tasks_.size());
   DCHECK_EQ(0u, completed_tasks_.size());
-  DCHECK_EQ(0u, running_task_count_);
 }
 
 void WorkerPool::Inner::Shutdown() {
   {
     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_pending_tasks_cv_.Signal();
   }
 
   while (workers_.size()) {
     scoped_ptr<base::DelegateSimpleThread> worker = workers_.take_front();
     worker->Join();
   }
+
+  // Cancel any pending OnIdle callback.
+  weak_ptr_factory_.InvalidateWeakPtrs();
 }
 
-void WorkerPool::Inner::PostTask(scoped_ptr<internal::WorkerPoolTask> task) {
+void WorkerPool::Inner::ScheduleTasks(
+    scoped_ptr<internal::WorkerPoolTaskGraph> task_graph) {
   base::AutoLock lock(lock_);
 
-  pending_tasks_.push_back(task.Pass());
+  // Move tasks not present in new graph to |completed_tasks_|.
+  while (task_graph_->HasMoreTasks()) {
+    scoped_refptr<internal::WorkerPoolTask> task = task_graph_->TakeTask(
+        task_graph_->TopTask());
+
+    // Task has not completed if present in new graph.
+    if (task_graph->HasTask(task))
+      continue;
+
+    completed_tasks_.push_back(task);
+  }
+
+  // Take any running tasks from new graph.
+  for (TaskSet::iterator it = running_tasks_.begin();
+       it != running_tasks_.end(); ++it) {
+    if (task_graph->HasTask(*it))
+      task_graph->TakeTask(*it);
+  }
+
+  // And take any completed tasks from new graph.
+  for (TaskDeque::iterator it = completed_tasks_.begin();
+       it != completed_tasks_.end(); ++it) {
+    if (task_graph->HasTask(*it))
+      task_graph->TakeTask(*it);
+  }
+
+  // Finally switch to the new graph.
+  task_graph_ = task_graph.Pass();
 
   // There is more work available, so wake up worker thread.
   has_pending_tasks_cv_.Signal();
 }
 
-bool WorkerPool::Inner::CollectCompletedTasks() {
+bool WorkerPool::Inner::CollectCompletedTasks(TaskDeque* completed_tasks) {
   base::AutoLock lock(lock_);
 
-  return AppendCompletedTasksWithLockAcquired(
-      &worker_pool_on_origin_thread_->completed_tasks_);
+  return CollectCompletedTasksWithLockAcquired(completed_tasks);
 }
 
-bool WorkerPool::Inner::AppendCompletedTasksWithLockAcquired(
-    ScopedPtrDeque<internal::WorkerPoolTask>* completed_tasks) {
+bool WorkerPool::Inner::CollectCompletedTasksWithLockAcquired(
+    TaskDeque* completed_tasks) {
   lock_.AssertAcquired();
 
-  while (completed_tasks_.size())
-    completed_tasks->push_back(completed_tasks_.take_front().Pass());
+  DCHECK_EQ(0u, completed_tasks->size());
+  completed_tasks->swap(completed_tasks_);
 
-  return !running_task_count_ && pending_tasks_.empty();
+  return running_tasks_.empty() && !task_graph_->HasMoreTasks();
 }
 
 void WorkerPool::Inner::ScheduleOnIdleWithLockAcquired() {
   lock_.AssertAcquired();
 
   if (on_idle_pending_)
     return;
   origin_loop_->PostTask(FROM_HERE, on_idle_callback_);
   on_idle_pending_ = true;
 }
 
 void WorkerPool::Inner::OnIdleOnOriginThread() {
+  TaskDeque completed_tasks;
+
   {
     base::AutoLock lock(lock_);
 
     DCHECK(on_idle_pending_);
     on_idle_pending_ = false;
 
     // Early out if no longer idle.
-    if (running_task_count_ || !pending_tasks_.empty())
+    if (!running_tasks_.empty() || task_graph_->HasMoreTasks())
       return;
 
-    AppendCompletedTasksWithLockAcquired(
-        &worker_pool_on_origin_thread_->completed_tasks_);
+    CollectCompletedTasksWithLockAcquired(&completed_tasks);
   }
 
-  worker_pool_on_origin_thread_->OnIdle();
+  worker_pool_on_origin_thread_->OnIdle(&completed_tasks);
 }
 
 void WorkerPool::Inner::Run() {
 #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
 
   base::AutoLock lock(lock_);
 
   // Get a unique thread index.
   int thread_index = next_thread_index_++;
 
   while (true) {
-    if (pending_tasks_.empty()) {
+    if (!task_graph_->HasMoreTasks()) {
       // Exit when shutdown is set and no more tasks are pending.
       if (shutdown_)

[vmpstr, 2013/05/06 23:08:48]

Now that task_graph_ contains all work that needs to be done, with just one
raster thread if we call Shutdown, it would still finish doing all the work
first (the enclosing if would fail). Is this by design? 

[reveman, 2013/05/07 01:33:54]

Yes, we could cancel all pending tasks at shutdown but the way to do that is by
scheduling an empty task graph as we still need to perform all completion
callbacks.

I'd like to keep the WorkerPool class generic and as it's trivial for the client
to do this I'd rather not have it be automatically done by
WorkerPool::Shutdown().

Cancel of all tasks is the prefered behavior for the tile manager though. Latest
patch takes care of this automatically in RasterWorkerPool::Shutdown().

         break;
 
-      // Schedule an idle callback if requested and not pending.
-      if (!running_task_count_)
+      // Schedule an idle callback if not tasks are running.

[vmpstr, 2013/05/06 23:08:48]

nit: not -> no

[reveman, 2013/05/07 01:33:54]

Done.

+      if (running_tasks_.empty())
         ScheduleOnIdleWithLockAcquired();
 
-      // Wait for new pending tasks.
+      // Wait for more tasks.
       has_pending_tasks_cv_.Wait();
       continue;
     }
 
-    // Get next task.
-    scoped_ptr<internal::WorkerPoolTask> task = pending_tasks_.take_front();
+    // Take the highest priority task from the task graph.
+    scoped_refptr<internal::WorkerPoolTask> task = task_graph_->TakeTask(
+        task_graph_->TopTask());
 
-    // Increment |running_task_count_| before starting to run task.
-    running_task_count_++;
+    // Insert task in |running_tasks_| before starting to run it.
+    running_tasks_.insert(task);
 
-    // There may be more work available, so wake up another
-    // worker thread.
+    // There may be more work available, so wake up another worker thread.
     has_pending_tasks_cv_.Signal();
 
     {
       base::AutoUnlock unlock(lock_);
 
       task->RunOnThread(thread_index);
     }
 
-    completed_tasks_.push_back(task.Pass());
+    // Remove task from |running_tasks_| now that we are done running it.
+    running_tasks_.erase(task);
 
-    // Decrement |running_task_count_| now that we are done running task.
-    running_task_count_--;
+    // Finally add task to |completed_tasks_|.
+    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_pending_tasks_cv_.Signal();
 }
 
 WorkerPool::WorkerPool(WorkerPoolClient* client,
                        size_t num_threads,
                        base::TimeDelta check_for_completed_tasks_delay,
                        const std::string& thread_name_prefix)
     : client_(client),
       origin_loop_(base::MessageLoopProxy::current()),
-      weak_ptr_factory_(this),
       check_for_completed_tasks_delay_(check_for_completed_tasks_delay),
       check_for_completed_tasks_pending_(false),
       inner_(make_scoped_ptr(new Inner(this,
                                        num_threads,
                                        thread_name_prefix))) {
 }
 
 WorkerPool::~WorkerPool() {
-  // Cancel all pending callbacks.
-  weak_ptr_factory_.InvalidateWeakPtrs();
-
-  DCHECK_EQ(0u, completed_tasks_.size());
 }
 
 void WorkerPool::Shutdown() {
   inner_->Shutdown();
-  inner_->CollectCompletedTasks();
-  DispatchCompletionCallbacks();
+
+  TaskDeque completed_tasks;
+  inner_->CollectCompletedTasks(&completed_tasks);
+  DispatchCompletionCallbacks(&completed_tasks);
 }
 
-void WorkerPool::PostTaskAndReply(
-    const Callback& task, const base::Closure& reply) {
-  PostTask(make_scoped_ptr(new WorkerPoolTaskImpl(
-                               task,
-                               reply)).PassAs<internal::WorkerPoolTask>());
-}
-
-void WorkerPool::OnIdle() {
+void WorkerPool::OnIdle(TaskDeque* completed_tasks) {
   TRACE_EVENT0("cc", "WorkerPool::OnIdle");
 
-  DispatchCompletionCallbacks();
+  DispatchCompletionCallbacks(completed_tasks);
+
+  // Cancel any pending check for completed tasks.
+  check_for_completed_tasks_callback_.Cancel();
+  check_for_completed_tasks_pending_ = false;
 }
 
 void WorkerPool::ScheduleCheckForCompletedTasks() {
   if (check_for_completed_tasks_pending_)
     return;
+  check_for_completed_tasks_callback_.Reset(
+      base::Bind(&WorkerPool::CheckForCompletedTasks,
+                 base::Unretained(this)));
   origin_loop_->PostDelayedTask(
       FROM_HERE,
-      base::Bind(&WorkerPool::CheckForCompletedTasks,
-                 weak_ptr_factory_.GetWeakPtr()),
+      check_for_completed_tasks_callback_.callback(),
       check_for_completed_tasks_delay_);
   check_for_completed_tasks_pending_ = true;
 }
 
 void WorkerPool::CheckForCompletedTasks() {
   TRACE_EVENT0("cc", "WorkerPool::CheckForCompletedTasks");
   DCHECK(check_for_completed_tasks_pending_);
   check_for_completed_tasks_pending_ = false;
 
+  TaskDeque completed_tasks;
+
   // Schedule another check for completed tasks if not idle.
-  if (!inner_->CollectCompletedTasks())
+  if (!inner_->CollectCompletedTasks(&completed_tasks))
     ScheduleCheckForCompletedTasks();
 
-  DispatchCompletionCallbacks();
+  DispatchCompletionCallbacks(&completed_tasks);
 }
 
-void WorkerPool::DispatchCompletionCallbacks() {
+void WorkerPool::DispatchCompletionCallbacks(TaskDeque* completed_tasks) {
   TRACE_EVENT0("cc", "WorkerPool::DispatchCompletionCallbacks");
 
-  if (completed_tasks_.empty())
+  // Early out when |completed_tasks| is empty to prevent unnecessary
+  // call to DidFinishDispatchingWorkerPoolCompletionCallbacks().
+  if (completed_tasks->empty())
     return;
 
-  while (completed_tasks_.size()) {
-    scoped_ptr<internal::WorkerPoolTask> task = completed_tasks_.take_front();
-    task->DidComplete();
+  while (completed_tasks->size()) {

[vmpstr, 2013/05/06 23:08:48]

nit: consider !empty

[reveman, 2013/05/07 01:33:54]

Done.

+    scoped_refptr<internal::WorkerPoolTask> task = completed_tasks->front();
+    completed_tasks->pop_front();
+    task->DispatchCompletionCallback();
   }
 
   client_->DidFinishDispatchingWorkerPoolCompletionCallbacks();
 }
 
-void WorkerPool::PostTask(scoped_ptr<internal::WorkerPoolTask> task) {
-  // Schedule check for completed tasks if not pending.
-  ScheduleCheckForCompletedTasks();
-
-  inner_->PostTask(task.Pass());
+void WorkerPool::ScheduleTasks(
+    scoped_ptr<internal::WorkerPoolTaskGraph> task_graph) {
+  if (task_graph->HasMoreTasks())
+    ScheduleCheckForCompletedTasks();
+  inner_->ScheduleTasks(task_graph.Pass());
 }
 
 }  // namespace cc