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