cc/raster/task_graph_runner.cc

   1 // Copyright 2014 The Chromium Authors. All rights reserved.
   2 // Use of this source code is governed by a BSD-style license that can be
   3 // found in the LICENSE file.
   4
   5 #include "cc/raster/task_graph_runner.h"
   6
   7 #include <algorithm>
   8
   9 #include "base/containers/hash_tables.h"
  10 #include "base/strings/stringprintf.h"
  11 #include "base/threading/thread_restrictions.h"
  12 #include "base/trace_event/trace_event.h"
  13
  14 namespace cc {
  15 namespace {
  16
  17 // Helper class for iterating over all dependents of a task.
  18 class DependentIterator {
  19  public:
  20   DependentIterator(TaskGraph* graph, const Task* task)
  21       : graph_(graph),
  22         task_(task),
  23         current_index_(static_cast<size_t>(-1)),
  24         current_node_(NULL) {
  25     ++(*this);
  26   }
  27
  28   TaskGraph::Node& operator->() const {
  29     DCHECK_LT(current_index_, graph_->edges.size());
  30     DCHECK_EQ(graph_->edges[current_index_].task, task_);
  31     DCHECK(current_node_);
  32     return *current_node_;
  33   }
  34
  35   TaskGraph::Node& operator*() const {
  36     DCHECK_LT(current_index_, graph_->edges.size());
  37     DCHECK_EQ(graph_->edges[current_index_].task, task_);
  38     DCHECK(current_node_);
  39     return *current_node_;
  40   }
  41
  42   // Note: Performance can be improved by keeping edges sorted.
  43   DependentIterator& operator++() {
  44     // Find next dependency edge for |task_|.
  45     do {
  46       ++current_index_;
  47       if (current_index_ == graph_->edges.size())
  48         return *this;
  49     } while (graph_->edges[current_index_].task != task_);
  50
  51     // Now find the node for the dependent of this edge.
  52     TaskGraph::Node::Vector::iterator it =
  53         std::find_if(graph_->nodes.begin(),
  54                      graph_->nodes.end(),
  55                      TaskGraph::Node::TaskComparator(
  56                          graph_->edges[current_index_].dependent));
  57     DCHECK(it != graph_->nodes.end());
  58     current_node_ = &(*it);
  59
  60     return *this;
  61   }
  62
  63   operator bool() const { return current_index_ < graph_->edges.size(); }
  64
  65  private:
  66   TaskGraph* graph_;
  67   const Task* task_;
  68   size_t current_index_;
  69   TaskGraph::Node* current_node_;
  70 };
  71
  72 bool DependencyMismatch(const TaskGraph* graph) {
  73   // Value storage will be 0-initialized.
  74   base::hash_map<const Task*, size_t> dependents;
  75   for (const TaskGraph::Edge& edge : graph->edges)
  76     dependents[edge.dependent]++;
  77
  78   for (const TaskGraph::Node& node : graph->nodes) {
  79     if (dependents[node.task] != node.dependencies)
  80       return true;
  81   }
  82
  83   return false;
  84 }
  85
  86 }  // namespace
  87
  88 Task::Task() : will_run_(false), did_run_(false) {
  89 }
  90
  91 Task::~Task() {
  92   DCHECK(!will_run_);
  93 }
  94
  95 void Task::WillRun() {
  96   DCHECK(!will_run_);
  97   DCHECK(!did_run_);
  98   will_run_ = true;
  99 }
 100
 101 void Task::DidRun() {
 102   DCHECK(will_run_);
 103   will_run_ = false;
 104   did_run_ = true;
 105 }
 106
 107 bool Task::HasFinishedRunning() const { return did_run_; }
 108
 109 TaskGraph::TaskGraph() {}
 110
 111 TaskGraph::~TaskGraph() {}
 112
 113 void TaskGraph::Swap(TaskGraph* other) {
 114   nodes.swap(other->nodes);
 115   edges.swap(other->edges);
 116 }
 117
 118 void TaskGraph::Reset() {
 119   nodes.clear();
 120   edges.clear();
 121 }
 122
 123 TaskGraphRunner::TaskNamespace::TaskNamespace() {}
 124
 125 TaskGraphRunner::TaskNamespace::~TaskNamespace() {}
 126
 127 TaskGraphRunner::TaskGraphRunner()
 128     : lock_(),
 129       has_ready_to_run_tasks_cv_(&lock_),
 130       has_namespaces_with_finished_running_tasks_cv_(&lock_),
 131       next_namespace_id_(1),
 132       shutdown_(false) {}
 133
 134 TaskGraphRunner::~TaskGraphRunner() {
 135   {
 136     base::AutoLock lock(lock_);
 137
 138     DCHECK_EQ(0u, ready_to_run_namespaces_.size());
 139     DCHECK_EQ(0u, namespaces_.size());
 140   }
 141 }
 142
 143 NamespaceToken TaskGraphRunner::GetNamespaceToken() {
 144   base::AutoLock lock(lock_);
 145
 146   NamespaceToken token(next_namespace_id_++);
 147   DCHECK(namespaces_.find(token.id_) == namespaces_.end());
 148   return token;
 149 }
 150
 151 void TaskGraphRunner::ScheduleTasks(NamespaceToken token, TaskGraph* graph) {
 152   TRACE_EVENT2("cc",
 153                "TaskGraphRunner::ScheduleTasks",
 154                "num_nodes",
 155                graph->nodes.size(),
 156                "num_edges",
 157                graph->edges.size());
 158
 159   DCHECK(token.IsValid());
 160   DCHECK(!DependencyMismatch(graph));
 161
 162   {
 163     base::AutoLock lock(lock_);
 164
 165     DCHECK(!shutdown_);
 166
 167     TaskNamespace& task_namespace = namespaces_[token.id_];
 168
 169     // First adjust number of dependencies to reflect completed tasks.
 170     for (Task::Vector::iterator it = task_namespace.completed_tasks.begin();
 171          it != task_namespace.completed_tasks.end();
 172          ++it) {
 173       for (DependentIterator node_it(graph, it->get()); node_it; ++node_it) {
 174         TaskGraph::Node& node = *node_it;
 175         DCHECK_LT(0u, node.dependencies);
 176         node.dependencies--;
 177       }
 178     }
 179
 180     // Build new "ready to run" queue and remove nodes from old graph.
 181     task_namespace.ready_to_run_tasks.clear();
 182     for (TaskGraph::Node::Vector::iterator it = graph->nodes.begin();
 183          it != graph->nodes.end();
 184          ++it) {
 185       TaskGraph::Node& node = *it;
 186
 187       // Remove any old nodes that are associated with this task. The result is
 188       // that the old graph is left with all nodes not present in this graph,
 189       // which we use below to determine what tasks need to be canceled.
 190       TaskGraph::Node::Vector::iterator old_it =
 191           std::find_if(task_namespace.graph.nodes.begin(),
 192                        task_namespace.graph.nodes.end(),
 193                        TaskGraph::Node::TaskComparator(node.task));
 194       if (old_it != task_namespace.graph.nodes.end()) {
 195         std::swap(*old_it, task_namespace.graph.nodes.back());
 196         task_namespace.graph.nodes.pop_back();
 197       }
 198
 199       // Task is not ready to run if dependencies are not yet satisfied.
 200       if (node.dependencies)
 201         continue;
 202
 203       // Skip if already finished running task.
 204       if (node.task->HasFinishedRunning())
 205         continue;
 206
 207       // Skip if already running.
 208       if (std::find(task_namespace.running_tasks.begin(),
 209                     task_namespace.running_tasks.end(),
 210                     node.task) != task_namespace.running_tasks.end())
 211         continue;
 212
 213       task_namespace.ready_to_run_tasks.push_back(
 214           PrioritizedTask(node.task, node.priority));
 215     }
 216
 217     // Rearrange the elements in |ready_to_run_tasks| in such a way that they
 218     // form a heap.
 219     std::make_heap(task_namespace.ready_to_run_tasks.begin(),
 220                    task_namespace.ready_to_run_tasks.end(),
 221                    CompareTaskPriority);
 222
 223     // Swap task graph.
 224     task_namespace.graph.Swap(graph);
 225
 226     // Determine what tasks in old graph need to be canceled.
 227     for (TaskGraph::Node::Vector::iterator it = graph->nodes.begin();
 228          it != graph->nodes.end();
 229          ++it) {
 230       TaskGraph::Node& node = *it;
 231
 232       // Skip if already finished running task.
 233       if (node.task->HasFinishedRunning())
 234         continue;
 235
 236       // Skip if already running.
 237       if (std::find(task_namespace.running_tasks.begin(),
 238                     task_namespace.running_tasks.end(),
 239                     node.task) != task_namespace.running_tasks.end())
 240         continue;
 241
 242       DCHECK(std::find(task_namespace.completed_tasks.begin(),
 243                        task_namespace.completed_tasks.end(),
 244                        node.task) == task_namespace.completed_tasks.end());
 245       task_namespace.completed_tasks.push_back(node.task);
 246     }
 247
 248     // Build new "ready to run" task namespaces queue.
 249     ready_to_run_namespaces_.clear();
 250     for (TaskNamespaceMap::iterator it = namespaces_.begin();
 251          it != namespaces_.end();
 252          ++it) {
 253       if (!it->second.ready_to_run_tasks.empty())
 254         ready_to_run_namespaces_.push_back(&it->second);
 255     }
 256
 257     // Rearrange the task namespaces in |ready_to_run_namespaces_| in such a way
 258     // that they form a heap.
 259     std::make_heap(ready_to_run_namespaces_.begin(),
 260                    ready_to_run_namespaces_.end(),
 261                    CompareTaskNamespacePriority);
 262
 263     // If there is more work available, wake up worker thread.
 264     if (!ready_to_run_namespaces_.empty())
 265       has_ready_to_run_tasks_cv_.Signal();
 266   }
 267 }
 268
 269 void TaskGraphRunner::WaitForTasksToFinishRunning(NamespaceToken token) {
 270   TRACE_EVENT0("cc", "TaskGraphRunner::WaitForTasksToFinishRunning");
 271
 272   DCHECK(token.IsValid());
 273
 274   {
 275     base::AutoLock lock(lock_);
 276     base::ThreadRestrictions::ScopedAllowWait allow_wait;
 277
 278     TaskNamespaceMap::const_iterator it = namespaces_.find(token.id_);
 279     if (it == namespaces_.end())
 280       return;
 281
 282     const TaskNamespace& task_namespace = it->second;
 283
 284     while (!HasFinishedRunningTasksInNamespace(&task_namespace))
 285       has_namespaces_with_finished_running_tasks_cv_.Wait();
 286
 287     // There may be other namespaces that have finished running tasks, so wake
 288     // up another origin thread.
 289     has_namespaces_with_finished_running_tasks_cv_.Signal();
 290   }
 291 }
 292
 293 void TaskGraphRunner::CollectCompletedTasks(NamespaceToken token,
 294                                             Task::Vector* completed_tasks) {
 295   TRACE_EVENT0("cc", "TaskGraphRunner::CollectCompletedTasks");
 296
 297   DCHECK(token.IsValid());
 298
 299   {
 300     base::AutoLock lock(lock_);
 301
 302     TaskNamespaceMap::iterator it = namespaces_.find(token.id_);
 303     if (it == namespaces_.end())
 304       return;
 305
 306     TaskNamespace& task_namespace = it->second;
 307
 308     DCHECK_EQ(0u, completed_tasks->size());
 309     completed_tasks->swap(task_namespace.completed_tasks);
 310     if (!HasFinishedRunningTasksInNamespace(&task_namespace))
 311       return;
 312
 313     // Remove namespace if finished running tasks.
 314     DCHECK_EQ(0u, task_namespace.completed_tasks.size());
 315     DCHECK_EQ(0u, task_namespace.ready_to_run_tasks.size());
 316     DCHECK_EQ(0u, task_namespace.running_tasks.size());
 317     namespaces_.erase(it);
 318   }
 319 }
 320
 321 void TaskGraphRunner::Shutdown() {
 322   base::AutoLock lock(lock_);
 323
 324   DCHECK_EQ(0u, ready_to_run_namespaces_.size());
 325   DCHECK_EQ(0u, namespaces_.size());
 326
 327   DCHECK(!shutdown_);
 328   shutdown_ = true;
 329
 330   // Wake up a worker so it knows it should exit. This will cause all workers
 331   // to exit as each will wake up another worker before exiting.
 332   has_ready_to_run_tasks_cv_.Signal();
 333 }
 334
 335 void TaskGraphRunner::FlushForTesting() {
 336   base::AutoLock lock(lock_);
 337
 338   while (std::find_if(namespaces_.begin(), namespaces_.end(),
 339                       [](const TaskNamespaceMap::value_type& entry) {
 340                         return !HasFinishedRunningTasksInNamespace(
 341                             &entry.second);
 342                       }) != namespaces_.end()) {
 343     has_namespaces_with_finished_running_tasks_cv_.Wait();
 344   }
 345 }
 346
 347 void TaskGraphRunner::Run() {
 348   base::AutoLock lock(lock_);
 349
 350   while (true) {
 351     if (ready_to_run_namespaces_.empty()) {
 352       // Exit when shutdown is set and no more tasks are pending.
 353       if (shutdown_)
 354         break;
 355
 356       // Wait for more tasks.
 357       has_ready_to_run_tasks_cv_.Wait();
 358       continue;
 359     }
 360
 361     RunTaskWithLockAcquired();
 362   }
 363
 364   // We noticed we should exit. Wake up the next worker so it knows it should
 365   // exit as well (because the Shutdown() code only signals once).
 366   has_ready_to_run_tasks_cv_.Signal();
 367 }
 368
 369 void TaskGraphRunner::RunUntilIdle() {
 370   base::AutoLock lock(lock_);
 371
 372   while (!ready_to_run_namespaces_.empty())
 373     RunTaskWithLockAcquired();
 374 }
 375
 376 void TaskGraphRunner::RunTaskWithLockAcquired() {
 377   TRACE_EVENT0("toplevel", "TaskGraphRunner::RunTask");
 378
 379   lock_.AssertAcquired();
 380   DCHECK(!ready_to_run_namespaces_.empty());
 381
 382   // Take top priority TaskNamespace from |ready_to_run_namespaces_|.
 383   std::pop_heap(ready_to_run_namespaces_.begin(),
 384                 ready_to_run_namespaces_.end(),
 385                 CompareTaskNamespacePriority);
 386   TaskNamespace* task_namespace = ready_to_run_namespaces_.back();
 387   ready_to_run_namespaces_.pop_back();
 388   DCHECK(!task_namespace->ready_to_run_tasks.empty());
 389
 390   // Take top priority task from |ready_to_run_tasks|.
 391   std::pop_heap(task_namespace->ready_to_run_tasks.begin(),
 392                 task_namespace->ready_to_run_tasks.end(),
 393                 CompareTaskPriority);
 394   scoped_refptr<Task> task(task_namespace->ready_to_run_tasks.back().task);
 395   task_namespace->ready_to_run_tasks.pop_back();
 396
 397   // Add task namespace back to |ready_to_run_namespaces_| if not empty after
 398   // taking top priority task.
 399   if (!task_namespace->ready_to_run_tasks.empty()) {
 400     ready_to_run_namespaces_.push_back(task_namespace);
 401     std::push_heap(ready_to_run_namespaces_.begin(),
 402                    ready_to_run_namespaces_.end(),
 403                    CompareTaskNamespacePriority);
 404   }
 405
 406   // Add task to |running_tasks|.
 407   task_namespace->running_tasks.push_back(task.get());
 408
 409   // There may be more work available, so wake up another worker thread.
 410   has_ready_to_run_tasks_cv_.Signal();
 411
 412   // Call WillRun() before releasing |lock_| and running task.
 413   task->WillRun();
 414
 415   {
 416     base::AutoUnlock unlock(lock_);
 417
 418     task->RunOnWorkerThread();
 419   }
 420
 421   // This will mark task as finished running.
 422   task->DidRun();
 423
 424   // Remove task from |running_tasks|.
 425   TaskVector::iterator it = std::find(task_namespace->running_tasks.begin(),
 426                                       task_namespace->running_tasks.end(),
 427                                       task.get());
 428   DCHECK(it != task_namespace->running_tasks.end());
 429   std::swap(*it, task_namespace->running_tasks.back());
 430   task_namespace->running_tasks.pop_back();
 431
 432   // Now iterate over all dependents to decrement dependencies and check if they
 433   // are ready to run.
 434   bool ready_to_run_namespaces_has_heap_properties = true;
 435   for (DependentIterator it(&task_namespace->graph, task.get()); it; ++it) {
 436     TaskGraph::Node& dependent_node = *it;
 437
 438     DCHECK_LT(0u, dependent_node.dependencies);
 439     dependent_node.dependencies--;
 440     // Task is ready if it has no dependencies. Add it to |ready_to_run_tasks_|.
 441     if (!dependent_node.dependencies) {
 442       bool was_empty = task_namespace->ready_to_run_tasks.empty();
 443       task_namespace->ready_to_run_tasks.push_back(
 444           PrioritizedTask(dependent_node.task, dependent_node.priority));
 445       std::push_heap(task_namespace->ready_to_run_tasks.begin(),
 446                      task_namespace->ready_to_run_tasks.end(),
 447                      CompareTaskPriority);
 448       // Task namespace is ready if it has at least one ready to run task. Add
 449       // it to |ready_to_run_namespaces_| if it just become ready.
 450       if (was_empty) {
 451         DCHECK(std::find(ready_to_run_namespaces_.begin(),
 452                          ready_to_run_namespaces_.end(),
 453                          task_namespace) == ready_to_run_namespaces_.end());
 454         ready_to_run_namespaces_.push_back(task_namespace);
 455       }
 456       ready_to_run_namespaces_has_heap_properties = false;
 457     }
 458   }
 459
 460   // Rearrange the task namespaces in |ready_to_run_namespaces_| in such a way
 461   // that they yet again form a heap.
 462   if (!ready_to_run_namespaces_has_heap_properties) {
 463     std::make_heap(ready_to_run_namespaces_.begin(),
 464                    ready_to_run_namespaces_.end(),
 465                    CompareTaskNamespacePriority);
 466   }
 467
 468   // Finally add task to |completed_tasks_|.
 469   task_namespace->completed_tasks.push_back(task);
 470
 471   // If namespace has finished running all tasks, wake up origin thread.
 472   if (HasFinishedRunningTasksInNamespace(task_namespace))
 473     has_namespaces_with_finished_running_tasks_cv_.Signal();
 474 }
 475
 476 }  // namespace cc