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