OLD | NEW |
1 // Copyright 2013 The Chromium Authors. All rights reserved. | 1 // Copyright 2013 The Chromium Authors. All rights reserved. |
2 // Use of this source code is governed by a BSD-style license that can be | 2 // Use of this source code is governed by a BSD-style license that can be |
3 // found in the LICENSE file. | 3 // found in the LICENSE file. |
4 | 4 |
5 #include "cc/resources/worker_pool.h" | 5 #include "cc/resources/worker_pool.h" |
6 | 6 |
7 #include <algorithm> | 7 #include <algorithm> |
8 #include <queue> | 8 #include <queue> |
9 | 9 |
10 #include "base/bind.h" | 10 #include "base/bind.h" |
| 11 #include "base/command_line.h" |
11 #include "base/containers/hash_tables.h" | 12 #include "base/containers/hash_tables.h" |
12 #include "base/debug/trace_event.h" | 13 #include "base/debug/trace_event.h" |
| 14 #include "base/lazy_instance.h" |
| 15 #include "base/memory/linked_ptr.h" |
13 #include "base/strings/stringprintf.h" | 16 #include "base/strings/stringprintf.h" |
14 #include "base/synchronization/condition_variable.h" | 17 #include "base/synchronization/condition_variable.h" |
15 #include "base/threading/simple_thread.h" | 18 #include "base/threading/simple_thread.h" |
16 #include "base/threading/thread_restrictions.h" | 19 #include "base/threading/thread_restrictions.h" |
17 #include "cc/base/scoped_ptr_deque.h" | 20 #include "cc/base/scoped_ptr_deque.h" |
| 21 #include "cc/base/switches.h" |
18 | 22 |
19 namespace cc { | 23 namespace cc { |
20 | 24 |
21 namespace internal { | 25 namespace { |
22 | 26 |
23 WorkerPoolTask::WorkerPoolTask() | 27 // TaskGraphRunners can process task graphs from multiple |
24 : did_schedule_(false), | 28 // workerpool instances. All members are guarded by |lock_|. |
25 did_run_(false), | 29 class TaskGraphRunner : public base::DelegateSimpleThread::Delegate { |
26 did_complete_(false) { | 30 public: |
27 } | 31 typedef WorkerPool::TaskGraph TaskGraph; |
| 32 typedef WorkerPool::TaskVector TaskVector; |
28 | 33 |
29 WorkerPoolTask::~WorkerPoolTask() { | 34 TaskGraphRunner(size_t num_threads, const std::string& thread_name_prefix); |
30 DCHECK_EQ(did_schedule_, did_complete_); | 35 virtual ~TaskGraphRunner(); |
31 DCHECK(!did_run_ || did_schedule_); | |
32 DCHECK(!did_run_ || did_complete_); | |
33 } | |
34 | 36 |
35 void WorkerPoolTask::DidSchedule() { | 37 void Register(const WorkerPool* worker_pool); |
36 DCHECK(!did_complete_); | 38 void Unregister(const WorkerPool* worker_pool); |
37 did_schedule_ = true; | 39 void WaitForTasksToFinishRunning(const WorkerPool* worker_pool); |
38 } | |
39 | |
40 void WorkerPoolTask::WillRun() { | |
41 DCHECK(did_schedule_); | |
42 DCHECK(!did_complete_); | |
43 DCHECK(!did_run_); | |
44 } | |
45 | |
46 void WorkerPoolTask::DidRun() { | |
47 did_run_ = true; | |
48 } | |
49 | |
50 void WorkerPoolTask::WillComplete() { | |
51 DCHECK(!did_complete_); | |
52 } | |
53 | |
54 void WorkerPoolTask::DidComplete() { | |
55 DCHECK(did_schedule_); | |
56 DCHECK(!did_complete_); | |
57 did_complete_ = true; | |
58 } | |
59 | |
60 bool WorkerPoolTask::HasFinishedRunning() const { | |
61 return did_run_; | |
62 } | |
63 | |
64 bool WorkerPoolTask::HasCompleted() const { | |
65 return did_complete_; | |
66 } | |
67 | |
68 GraphNode::GraphNode(internal::WorkerPoolTask* task, unsigned priority) | |
69 : task_(task), | |
70 priority_(priority), | |
71 num_dependencies_(0) { | |
72 } | |
73 | |
74 GraphNode::~GraphNode() { | |
75 } | |
76 | |
77 } // namespace internal | |
78 | |
79 // Internal to the worker pool. Any data or logic that needs to be | |
80 // shared between threads lives in this class. All members are guarded | |
81 // by |lock_|. | |
82 class WorkerPool::Inner : public base::DelegateSimpleThread::Delegate { | |
83 public: | |
84 Inner(size_t num_threads, const std::string& thread_name_prefix); | |
85 virtual ~Inner(); | |
86 | |
87 void Shutdown(); | |
88 | 40 |
89 // Schedule running of tasks in |graph|. Tasks previously scheduled but | 41 // Schedule running of tasks in |graph|. Tasks previously scheduled but |
90 // no longer needed will be canceled unless already running. Canceled | 42 // no longer needed will be canceled unless already running. Canceled |
91 // tasks are moved to |completed_tasks_| without being run. The result | 43 // tasks are moved to |completed_tasks| without being run. The result |
92 // is that once scheduled, a task is guaranteed to end up in the | 44 // is that once scheduled, a task is guaranteed to end up in the |
93 // |completed_tasks_| queue even if they later get canceled by another | 45 // |completed_tasks| queue even if it later get canceled by another |
94 // call to SetTaskGraph(). | 46 // call to SetTaskGraph(). |
95 void SetTaskGraph(TaskGraph* graph); | 47 void SetTaskGraph(const WorkerPool* worker_pool, TaskGraph* graph); |
96 | 48 |
97 // Collect all completed tasks in |completed_tasks|. | 49 // Collect all completed tasks in |completed_tasks|. |
98 void CollectCompletedTasks(TaskVector* completed_tasks); | 50 void CollectCompletedTasks(const WorkerPool* worker_pool, |
| 51 TaskVector* completed_tasks); |
99 | 52 |
100 private: | 53 private: |
101 class PriorityComparator { | 54 class TaskPriorityComparator { |
102 public: | 55 public: |
103 bool operator()(const internal::GraphNode* a, | 56 bool operator()(internal::GraphNode* a, |
104 const internal::GraphNode* b) { | 57 internal::GraphNode* b) { |
105 // In this system, numerically lower priority is run first. | 58 // In this system, numerically lower priority is run first. |
106 if (a->priority() != b->priority()) | 59 if (a->priority() != b->priority()) |
107 return a->priority() > b->priority(); | 60 return a->priority() > b->priority(); |
108 | 61 |
109 // Run task with most dependents first when priority is the same. | 62 // Run task with most dependents first when priority is the same. |
110 return a->dependents().size() < b->dependents().size(); | 63 return a->dependents().size() < b->dependents().size(); |
111 } | 64 } |
112 }; | 65 }; |
113 | 66 |
| 67 // Ordered set of tasks that are ready to run. |
| 68 typedef std::priority_queue<internal::GraphNode*, |
| 69 std::vector<internal::GraphNode*>, |
| 70 TaskPriorityComparator> TaskQueue; |
| 71 |
| 72 struct TaskNamespace { |
| 73 TaskGraph pending_tasks; |
| 74 TaskGraph running_tasks; |
| 75 TaskVector completed_tasks; |
| 76 TaskQueue ready_to_run_tasks; |
| 77 }; |
| 78 |
| 79 class TaskNamespacePriorityComparator { |
| 80 public: |
| 81 bool operator()(TaskNamespace* a, |
| 82 TaskNamespace* b) { |
| 83 return task_comparators_.operator()(a->ready_to_run_tasks.top(), |
| 84 b->ready_to_run_tasks.top()); |
| 85 } |
| 86 TaskPriorityComparator task_comparators_; |
| 87 }; |
| 88 |
| 89 typedef std::map<const WorkerPool*, linked_ptr<TaskNamespace> > |
| 90 TaskNamespaceMap; |
| 91 |
| 92 // Ordered set of tasks namespaces that have ready to run tasks. |
| 93 typedef std::priority_queue<TaskNamespace*, |
| 94 std::vector<TaskNamespace*>, |
| 95 TaskNamespacePriorityComparator> |
| 96 TaskNamespaceQueue; |
| 97 |
114 // Overridden from base::DelegateSimpleThread: | 98 // Overridden from base::DelegateSimpleThread: |
115 virtual void Run() OVERRIDE; | 99 virtual void Run() OVERRIDE; |
116 | 100 |
| 101 inline bool has_finished_running_tasks(TaskNamespace* task_namespace) { |
| 102 return (task_namespace->pending_tasks.empty() && |
| 103 task_namespace->running_tasks.empty()); |
| 104 } |
| 105 |
117 // This lock protects all members of this class except | 106 // This lock protects all members of this class except |
118 // |worker_pool_on_origin_thread_|. Do not read or modify anything | 107 // |worker_pool_on_origin_thread_|. Do not read or modify anything |
119 // without holding this lock. Do not block while holding this lock. | 108 // without holding this lock. Do not block while holding this lock. |
120 mutable base::Lock lock_; | 109 mutable base::Lock lock_; |
121 | 110 |
122 // Condition variable that is waited on by worker threads until new | 111 // Condition variable that is waited on by worker threads until new |
123 // tasks are ready to run or shutdown starts. | 112 // tasks are ready to run or shutdown starts. |
124 base::ConditionVariable has_ready_to_run_tasks_cv_; | 113 base::ConditionVariable has_ready_to_run_tasks_cv_; |
125 | 114 |
| 115 // Condition variable that is waited on by origin threads until a |
| 116 // namespace has finished running all associated tasks. |
| 117 base::ConditionVariable has_namespaces_with_finished_running_tasks_cv_; |
| 118 |
126 // Provides each running thread loop with a unique index. First thread | 119 // Provides each running thread loop with a unique index. First thread |
127 // loop index is 0. | 120 // loop index is 0. |
128 unsigned next_thread_index_; | 121 unsigned next_thread_index_; |
129 | 122 |
130 // Set during shutdown. Tells workers to exit when no more tasks | 123 // Set during shutdown. Tells workers to exit when no more tasks |
131 // are pending. | 124 // are pending. |
132 bool shutdown_; | 125 bool shutdown_; |
133 | 126 |
134 // This set contains all pending tasks. | |
135 GraphNodeMap pending_tasks_; | |
136 | |
137 // Ordered set of tasks that are ready to run. | |
138 typedef std::priority_queue<internal::GraphNode*, | |
139 std::vector<internal::GraphNode*>, | |
140 PriorityComparator> TaskQueue; | |
141 TaskQueue ready_to_run_tasks_; | |
142 | |
143 // This set contains all currently running tasks. | |
144 GraphNodeMap running_tasks_; | |
145 | |
146 // Completed tasks not yet collected by origin thread. | |
147 TaskVector completed_tasks_; | |
148 | |
149 ScopedPtrDeque<base::DelegateSimpleThread> workers_; | 127 ScopedPtrDeque<base::DelegateSimpleThread> workers_; |
150 | 128 |
151 DISALLOW_COPY_AND_ASSIGN(Inner); | 129 TaskNamespaceMap namespaces_; |
| 130 |
| 131 TaskNamespaceQueue ready_to_run_namespaces_; |
| 132 |
| 133 DISALLOW_COPY_AND_ASSIGN(TaskGraphRunner); |
152 }; | 134 }; |
153 | 135 |
154 WorkerPool::Inner::Inner( | 136 TaskGraphRunner::TaskGraphRunner( |
155 size_t num_threads, const std::string& thread_name_prefix) | 137 size_t num_threads, const std::string& thread_name_prefix) |
156 : lock_(), | 138 : lock_(), |
157 has_ready_to_run_tasks_cv_(&lock_), | 139 has_ready_to_run_tasks_cv_(&lock_), |
| 140 has_namespaces_with_finished_running_tasks_cv_(&lock_), |
158 next_thread_index_(0), | 141 next_thread_index_(0), |
159 shutdown_(false) { | 142 shutdown_(false) { |
160 base::AutoLock lock(lock_); | 143 base::AutoLock lock(lock_); |
161 | 144 |
162 while (workers_.size() < num_threads) { | 145 while (workers_.size() < num_threads) { |
163 scoped_ptr<base::DelegateSimpleThread> worker = make_scoped_ptr( | 146 scoped_ptr<base::DelegateSimpleThread> worker = make_scoped_ptr( |
164 new base::DelegateSimpleThread( | 147 new base::DelegateSimpleThread( |
165 this, | 148 this, |
166 thread_name_prefix + | 149 thread_name_prefix + |
167 base::StringPrintf( | 150 base::StringPrintf( |
168 "Worker%u", | 151 "Worker%u", |
169 static_cast<unsigned>(workers_.size() + 1)).c_str())); | 152 static_cast<unsigned>(workers_.size() + 1)).c_str())); |
170 worker->Start(); | 153 worker->Start(); |
171 #if defined(OS_ANDROID) || defined(OS_LINUX) | 154 #if defined(OS_ANDROID) || defined(OS_LINUX) |
172 worker->SetThreadPriority(base::kThreadPriority_Background); | 155 worker->SetThreadPriority(base::kThreadPriority_Background); |
173 #endif | 156 #endif |
174 workers_.push_back(worker.Pass()); | 157 workers_.push_back(worker.Pass()); |
175 } | 158 } |
176 } | 159 } |
177 | 160 |
178 WorkerPool::Inner::~Inner() { | 161 TaskGraphRunner::~TaskGraphRunner() { |
179 base::AutoLock lock(lock_); | |
180 | |
181 DCHECK(shutdown_); | |
182 | |
183 DCHECK_EQ(0u, pending_tasks_.size()); | |
184 DCHECK_EQ(0u, ready_to_run_tasks_.size()); | |
185 DCHECK_EQ(0u, running_tasks_.size()); | |
186 DCHECK_EQ(0u, completed_tasks_.size()); | |
187 } | |
188 | |
189 void WorkerPool::Inner::Shutdown() { | |
190 { | 162 { |
191 base::AutoLock lock(lock_); | 163 base::AutoLock lock(lock_); |
192 | 164 |
| 165 DCHECK_EQ(0u, ready_to_run_namespaces_.size()); |
| 166 DCHECK_EQ(0u, namespaces_.size()); |
| 167 |
193 DCHECK(!shutdown_); | 168 DCHECK(!shutdown_); |
194 shutdown_ = true; | 169 shutdown_ = true; |
195 | 170 |
196 // Wake up a worker so it knows it should exit. This will cause all workers | 171 // Wake up a worker so it knows it should exit. This will cause all workers |
197 // to exit as each will wake up another worker before exiting. | 172 // to exit as each will wake up another worker before exiting. |
198 has_ready_to_run_tasks_cv_.Signal(); | 173 has_ready_to_run_tasks_cv_.Signal(); |
199 } | 174 } |
200 | 175 |
201 while (workers_.size()) { | 176 while (workers_.size()) { |
202 scoped_ptr<base::DelegateSimpleThread> worker = workers_.take_front(); | 177 scoped_ptr<base::DelegateSimpleThread> worker = workers_.take_front(); |
203 // http://crbug.com/240453 - Join() is considered IO and will block this | 178 // http://crbug.com/240453 - Join() is considered IO and will block this |
204 // thread. See also http://crbug.com/239423 for further ideas. | 179 // thread. See also http://crbug.com/239423 for further ideas. |
205 base::ThreadRestrictions::ScopedAllowIO allow_io; | 180 base::ThreadRestrictions::ScopedAllowIO allow_io; |
206 worker->Join(); | 181 worker->Join(); |
207 } | 182 } |
208 } | 183 } |
209 | 184 |
210 void WorkerPool::Inner::SetTaskGraph(TaskGraph* graph) { | 185 void TaskGraphRunner::Register(const WorkerPool* worker_pool) { |
| 186 base::AutoLock lock(lock_); |
| 187 |
| 188 DCHECK(namespaces_.find(worker_pool) == namespaces_.end()); |
| 189 linked_ptr<TaskNamespace> task_set = make_linked_ptr(new TaskNamespace()); |
| 190 namespaces_[worker_pool] = task_set; |
| 191 } |
| 192 |
| 193 void TaskGraphRunner::Unregister(const WorkerPool* worker_pool) { |
| 194 base::AutoLock lock(lock_); |
| 195 |
| 196 DCHECK(namespaces_.find(worker_pool) != namespaces_.end()); |
| 197 |
| 198 TaskNamespace* task_namespace = namespaces_[worker_pool].get(); |
| 199 DCHECK_EQ(0u, task_namespace->pending_tasks.size()); |
| 200 DCHECK_EQ(0u, task_namespace->ready_to_run_tasks.size()); |
| 201 DCHECK_EQ(0u, task_namespace->running_tasks.size()); |
| 202 DCHECK_EQ(0u, task_namespace->completed_tasks.size()); |
| 203 |
| 204 namespaces_.erase(worker_pool); |
| 205 } |
| 206 |
| 207 void TaskGraphRunner::WaitForTasksToFinishRunning( |
| 208 const WorkerPool* worker_pool) { |
| 209 base::AutoLock lock(lock_); |
| 210 |
| 211 DCHECK(namespaces_.find(worker_pool) != namespaces_.end()); |
| 212 TaskNamespace* task_namespace = namespaces_[worker_pool].get(); |
| 213 |
| 214 while (true) { |
| 215 if (has_finished_running_tasks(task_namespace)) { |
| 216 // There maybe other namespaces that have finished running |
| 217 // running tasks, so wake up another origin thread |
| 218 has_namespaces_with_finished_running_tasks_cv_.Signal(); |
| 219 return; |
| 220 } |
| 221 has_namespaces_with_finished_running_tasks_cv_.Wait(); |
| 222 } |
| 223 } |
| 224 |
| 225 void TaskGraphRunner::SetTaskGraph(const WorkerPool* worker_pool, |
| 226 TaskGraph* graph) { |
211 // It is OK to call SetTaskGraph() after shutdown if |graph| is empty. | 227 // It is OK to call SetTaskGraph() after shutdown if |graph| is empty. |
212 DCHECK(graph->empty() || !shutdown_); | 228 DCHECK(graph->empty() || !shutdown_); |
213 | 229 |
214 GraphNodeMap new_pending_tasks; | 230 TaskGraph new_pending_tasks; |
215 GraphNodeMap new_running_tasks; | 231 TaskGraph new_running_tasks; |
216 TaskQueue new_ready_to_run_tasks; | 232 TaskQueue new_ready_to_run_tasks; |
| 233 TaskNamespaceQueue new_ready_to_run_namespaces; |
217 | 234 |
218 new_pending_tasks.swap(*graph); | 235 new_pending_tasks.swap(*graph); |
219 | 236 |
220 { | 237 { |
221 base::AutoLock lock(lock_); | 238 base::AutoLock lock(lock_); |
222 | 239 |
| 240 DCHECK(namespaces_.find(worker_pool) != namespaces_.end()); |
| 241 TaskNamespace* task_namespace = namespaces_[worker_pool].get(); |
| 242 |
223 // First remove all completed tasks from |new_pending_tasks| and | 243 // First remove all completed tasks from |new_pending_tasks| and |
224 // adjust number of dependencies. | 244 // adjust number of dependencies. |
225 for (TaskVector::iterator it = completed_tasks_.begin(); | 245 for (TaskVector::iterator it = task_namespace->completed_tasks.begin(); |
226 it != completed_tasks_.end(); ++it) { | 246 it != task_namespace->completed_tasks.end(); ++it) { |
227 internal::WorkerPoolTask* task = it->get(); | 247 internal::WorkerPoolTask* task = it->get(); |
228 | 248 |
229 scoped_ptr<internal::GraphNode> node = new_pending_tasks.take_and_erase( | 249 scoped_ptr<internal::GraphNode> node = new_pending_tasks.take_and_erase( |
230 task); | 250 task); |
231 if (node) { | 251 if (node) { |
232 for (internal::GraphNode::Vector::const_iterator it = | 252 for (internal::GraphNode::Vector::const_iterator it = |
233 node->dependents().begin(); | 253 node->dependents().begin(); |
234 it != node->dependents().end(); ++it) { | 254 it != node->dependents().end(); ++it) { |
235 internal::GraphNode* dependent_node = *it; | 255 internal::GraphNode* dependent_node = *it; |
236 dependent_node->remove_dependency(); | 256 dependent_node->remove_dependency(); |
237 } | 257 } |
238 } | 258 } |
239 } | 259 } |
240 | 260 |
241 // Build new running task set. | 261 // Build new running task set. |
242 for (GraphNodeMap::iterator it = running_tasks_.begin(); | 262 for (TaskGraph::iterator it = task_namespace->running_tasks.begin(); |
243 it != running_tasks_.end(); ++it) { | 263 it != task_namespace->running_tasks.end(); ++it) { |
244 internal::WorkerPoolTask* task = it->first; | 264 internal::WorkerPoolTask* task = it->first; |
245 // Transfer scheduled task value from |new_pending_tasks| to | 265 // Transfer scheduled task value from |new_pending_tasks| to |
246 // |new_running_tasks| if currently running. Value must be set to | 266 // |new_running_tasks| if currently running. Value must be set to |
247 // NULL if |new_pending_tasks| doesn't contain task. This does | 267 // NULL if |new_pending_tasks| doesn't contain task. This does |
248 // the right in both cases. | 268 // the right in both cases. |
249 new_running_tasks.set(task, new_pending_tasks.take_and_erase(task)); | 269 new_running_tasks.set(task, new_pending_tasks.take_and_erase(task)); |
250 } | 270 } |
251 | 271 |
252 // Build new "ready to run" tasks queue. | 272 // Build new "ready to run" tasks queue. |
253 // TODO(reveman): Create this queue when building the task graph instead. | 273 // TODO(reveman): Create this queue when building the task graph instead. |
254 for (GraphNodeMap::iterator it = new_pending_tasks.begin(); | 274 for (TaskGraph::iterator it = new_pending_tasks.begin(); |
255 it != new_pending_tasks.end(); ++it) { | 275 it != new_pending_tasks.end(); ++it) { |
256 internal::WorkerPoolTask* task = it->first; | 276 internal::WorkerPoolTask* task = it->first; |
257 DCHECK(task); | 277 DCHECK(task); |
258 internal::GraphNode* node = it->second; | 278 internal::GraphNode* node = it->second; |
259 | 279 |
260 // Completed tasks should not exist in |new_pending_tasks|. | 280 // Completed tasks should not exist in |new_pending_tasks|. |
261 DCHECK(!task->HasFinishedRunning()); | 281 DCHECK(!task->HasFinishedRunning()); |
262 | 282 |
263 // Call DidSchedule() to indicate that this task has been scheduled. | 283 // Call DidSchedule() to indicate that this task has been scheduled. |
264 // Note: This is only for debugging purposes. | 284 // Note: This is only for debugging purposes. |
265 task->DidSchedule(); | 285 task->DidSchedule(); |
266 | 286 |
267 if (!node->num_dependencies()) | 287 if (!node->num_dependencies()) |
268 new_ready_to_run_tasks.push(node); | 288 new_ready_to_run_tasks.push(node); |
269 | 289 |
270 // Erase the task from old pending tasks. | 290 // Erase the task from old pending tasks. |
271 pending_tasks_.erase(task); | 291 task_namespace->pending_tasks.erase(task); |
272 } | 292 } |
273 | 293 |
274 completed_tasks_.reserve(completed_tasks_.size() + pending_tasks_.size()); | 294 task_namespace->completed_tasks.reserve( |
| 295 task_namespace->completed_tasks.size() + |
| 296 task_namespace->pending_tasks.size()); |
275 | 297 |
276 // The items left in |pending_tasks_| need to be canceled. | 298 // The items left in |pending_tasks| need to be canceled. |
277 for (GraphNodeMap::const_iterator it = pending_tasks_.begin(); | 299 for (TaskGraph::const_iterator it = task_namespace->pending_tasks.begin(); |
278 it != pending_tasks_.end(); | 300 it != task_namespace->pending_tasks.end(); ++it) { |
279 ++it) { | 301 task_namespace->completed_tasks.push_back(it->first); |
280 completed_tasks_.push_back(it->first); | |
281 } | 302 } |
282 | 303 |
283 // Swap task sets. | 304 // Swap task sets. |
284 // Note: old tasks are intentionally destroyed after releasing |lock_|. | 305 // Note: old tasks are intentionally destroyed after releasing |lock_|. |
285 pending_tasks_.swap(new_pending_tasks); | 306 task_namespace->pending_tasks.swap(new_pending_tasks); |
286 running_tasks_.swap(new_running_tasks); | 307 task_namespace->running_tasks.swap(new_running_tasks); |
287 std::swap(ready_to_run_tasks_, new_ready_to_run_tasks); | 308 std::swap(task_namespace->ready_to_run_tasks, new_ready_to_run_tasks); |
288 | 309 |
289 // If |ready_to_run_tasks_| is empty, it means we either have | 310 // If |ready_to_run_tasks| is empty, it means we either have |
290 // running tasks, or we have no pending tasks. | 311 // running tasks, or we have no pending tasks. |
291 DCHECK(!ready_to_run_tasks_.empty() || | 312 DCHECK(!task_namespace->ready_to_run_tasks.empty() || |
292 (pending_tasks_.empty() || !running_tasks_.empty())); | 313 (task_namespace->pending_tasks.empty() || |
| 314 !task_namespace->running_tasks.empty())); |
| 315 |
| 316 // Build new "ready to run" task namespaces queue. |
| 317 for (TaskNamespaceMap::iterator it = namespaces_.begin(); |
| 318 it != namespaces_.end(); ++it) { |
| 319 if (!it->second->ready_to_run_tasks.empty()) |
| 320 new_ready_to_run_namespaces.push(it->second.get()); |
| 321 } |
| 322 std::swap(ready_to_run_namespaces_, new_ready_to_run_namespaces); |
293 | 323 |
294 // If there is more work available, wake up worker thread. | 324 // If there is more work available, wake up worker thread. |
295 if (!ready_to_run_tasks_.empty()) | 325 if (!ready_to_run_namespaces_.empty()) |
296 has_ready_to_run_tasks_cv_.Signal(); | 326 has_ready_to_run_tasks_cv_.Signal(); |
297 } | 327 } |
298 } | 328 } |
299 | 329 |
300 void WorkerPool::Inner::CollectCompletedTasks(TaskVector* completed_tasks) { | 330 void TaskGraphRunner::CollectCompletedTasks( |
| 331 const WorkerPool* worker_pool, TaskVector* completed_tasks) { |
301 base::AutoLock lock(lock_); | 332 base::AutoLock lock(lock_); |
302 | 333 |
303 DCHECK_EQ(0u, completed_tasks->size()); | 334 DCHECK_EQ(0u, completed_tasks->size()); |
304 completed_tasks->swap(completed_tasks_); | 335 DCHECK(namespaces_.find(worker_pool) != namespaces_.end()); |
| 336 completed_tasks->swap(namespaces_[worker_pool]->completed_tasks); |
305 } | 337 } |
306 | 338 |
307 void WorkerPool::Inner::Run() { | 339 void TaskGraphRunner::Run() { |
308 base::AutoLock lock(lock_); | 340 base::AutoLock lock(lock_); |
309 | 341 |
310 // Get a unique thread index. | 342 // Get a unique thread index. |
311 int thread_index = next_thread_index_++; | 343 int thread_index = next_thread_index_++; |
312 | 344 |
313 while (true) { | 345 while (true) { |
314 if (ready_to_run_tasks_.empty()) { | 346 if (ready_to_run_namespaces_.empty()) { |
315 // Exit when shutdown is set and no more tasks are pending. | 347 // Exit when shutdown is set and no more tasks are pending. |
316 if (shutdown_ && pending_tasks_.empty()) | 348 if (shutdown_) |
317 break; | 349 break; |
318 | 350 |
319 // Wait for more tasks. | 351 // Wait for more tasks. |
320 has_ready_to_run_tasks_cv_.Wait(); | 352 has_ready_to_run_tasks_cv_.Wait(); |
321 continue; | 353 continue; |
322 } | 354 } |
323 | 355 |
324 // Take top priority task from |ready_to_run_tasks_|. | 356 // Take top priority TaskNamespace from |ready_to_run_namespaces_|. |
| 357 TaskNamespace* task_namespace = ready_to_run_namespaces_.top(); |
| 358 ready_to_run_namespaces_.pop(); |
| 359 DCHECK(!task_namespace->ready_to_run_tasks.empty()); |
| 360 |
| 361 // Take top priority task from |ready_to_run_tasks|. |
325 scoped_refptr<internal::WorkerPoolTask> task( | 362 scoped_refptr<internal::WorkerPoolTask> task( |
326 ready_to_run_tasks_.top()->task()); | 363 task_namespace->ready_to_run_tasks.top()->task()); |
327 ready_to_run_tasks_.pop(); | 364 task_namespace->ready_to_run_tasks.pop(); |
328 | 365 |
329 // Move task from |pending_tasks_| to |running_tasks_|. | 366 // Add task namespace back to |ready_to_run_namespaces_| if not |
330 DCHECK(pending_tasks_.contains(task.get())); | 367 // empty after taking top priority task. |
331 DCHECK(!running_tasks_.contains(task.get())); | 368 if (!task_namespace->ready_to_run_tasks.empty()) |
332 running_tasks_.set(task.get(), pending_tasks_.take_and_erase(task.get())); | 369 ready_to_run_namespaces_.push(task_namespace); |
| 370 |
| 371 // Move task from |pending_tasks| to |running_tasks|. |
| 372 DCHECK(task_namespace->pending_tasks.contains(task.get())); |
| 373 DCHECK(!task_namespace->running_tasks.contains(task.get())); |
| 374 task_namespace->running_tasks.set( |
| 375 task.get(), |
| 376 task_namespace->pending_tasks.take_and_erase(task.get())); |
333 | 377 |
334 // There may be more work available, so wake up another worker thread. | 378 // There may be more work available, so wake up another worker thread. |
335 has_ready_to_run_tasks_cv_.Signal(); | 379 has_ready_to_run_tasks_cv_.Signal(); |
336 | 380 |
337 // Call WillRun() before releasing |lock_| and running task. | 381 // Call WillRun() before releasing |lock_| and running task. |
338 task->WillRun(); | 382 task->WillRun(); |
339 | 383 |
340 { | 384 { |
341 base::AutoUnlock unlock(lock_); | 385 base::AutoUnlock unlock(lock_); |
342 | 386 |
343 task->RunOnWorkerThread(thread_index); | 387 task->RunOnWorkerThread(thread_index); |
344 } | 388 } |
345 | 389 |
346 // This will mark task as finished running. | 390 // This will mark task as finished running. |
347 task->DidRun(); | 391 task->DidRun(); |
348 | 392 |
349 // Now iterate over all dependents to remove dependency and check | 393 // Now iterate over all dependents to remove dependency and check |
350 // if they are ready to run. | 394 // if they are ready to run. |
351 scoped_ptr<internal::GraphNode> node = running_tasks_.take_and_erase( | 395 scoped_ptr<internal::GraphNode> node = |
352 task.get()); | 396 task_namespace->running_tasks.take_and_erase(task.get()); |
353 if (node) { | 397 if (node) { |
354 for (internal::GraphNode::Vector::const_iterator it = | 398 for (internal::GraphNode::Vector::const_iterator it = |
355 node->dependents().begin(); | 399 node->dependents().begin(); |
356 it != node->dependents().end(); ++it) { | 400 it != node->dependents().end(); ++it) { |
357 internal::GraphNode* dependent_node = *it; | 401 internal::GraphNode* dependent_node = *it; |
358 | 402 |
359 dependent_node->remove_dependency(); | 403 dependent_node->remove_dependency(); |
360 // Task is ready if it has no dependencies. Add it to | 404 // Task is ready if it has no dependencies. Add it to |
361 // |ready_to_run_tasks_|. | 405 // |ready_to_run_tasks|. |
362 if (!dependent_node->num_dependencies()) | 406 if (!dependent_node->num_dependencies()) { |
363 ready_to_run_tasks_.push(dependent_node); | 407 bool was_empty = task_namespace->ready_to_run_tasks.empty(); |
| 408 task_namespace->ready_to_run_tasks.push(dependent_node); |
| 409 // Task namespace is ready if it has at least one ready |
| 410 // to run task. Add it to |ready_to_run_namespaces_| if |
| 411 // it just become ready. |
| 412 if (was_empty) |
| 413 ready_to_run_namespaces_.push(task_namespace); |
| 414 } |
364 } | 415 } |
365 } | 416 } |
366 | 417 |
367 // Finally add task to |completed_tasks_|. | 418 // Finally add task to |completed_tasks|. |
368 completed_tasks_.push_back(task); | 419 task_namespace->completed_tasks.push_back(task); |
| 420 |
| 421 // If namespace has finished running all tasks, wake up origin thread. |
| 422 if (has_finished_running_tasks(task_namespace)) |
| 423 has_namespaces_with_finished_running_tasks_cv_.Signal(); |
369 } | 424 } |
370 | 425 |
371 // We noticed we should exit. Wake up the next worker so it knows it should | 426 // We noticed we should exit. Wake up the next worker so it knows it should |
372 // exit as well (because the Shutdown() code only signals once). | 427 // exit as well (because the Shutdown() code only signals once). |
373 has_ready_to_run_tasks_cv_.Signal(); | 428 has_ready_to_run_tasks_cv_.Signal(); |
374 } | 429 } |
375 | 430 |
| 431 class CC_EXPORT CompositorRasterTaskGraphRunner |
| 432 : public TaskGraphRunner { |
| 433 public: |
| 434 CompositorRasterTaskGraphRunner() : TaskGraphRunner( |
| 435 switches::GetNumRasterThreads(), "CompositorRaster") { |
| 436 } |
| 437 }; |
| 438 |
| 439 base::LazyInstance<CompositorRasterTaskGraphRunner> |
| 440 g_task_graph_runner = LAZY_INSTANCE_INITIALIZER; |
| 441 |
| 442 } // namespace |
| 443 |
| 444 namespace internal { |
| 445 |
| 446 WorkerPoolTask::WorkerPoolTask() |
| 447 : did_schedule_(false), |
| 448 did_run_(false), |
| 449 did_complete_(false) { |
| 450 } |
| 451 |
| 452 WorkerPoolTask::~WorkerPoolTask() { |
| 453 DCHECK_EQ(did_schedule_, did_complete_); |
| 454 DCHECK(!did_run_ || did_schedule_); |
| 455 DCHECK(!did_run_ || did_complete_); |
| 456 } |
| 457 |
| 458 void WorkerPoolTask::DidSchedule() { |
| 459 DCHECK(!did_complete_); |
| 460 did_schedule_ = true; |
| 461 } |
| 462 |
| 463 void WorkerPoolTask::WillRun() { |
| 464 DCHECK(did_schedule_); |
| 465 DCHECK(!did_complete_); |
| 466 DCHECK(!did_run_); |
| 467 } |
| 468 |
| 469 void WorkerPoolTask::DidRun() { |
| 470 did_run_ = true; |
| 471 } |
| 472 |
| 473 void WorkerPoolTask::WillComplete() { |
| 474 DCHECK(!did_complete_); |
| 475 } |
| 476 |
| 477 void WorkerPoolTask::DidComplete() { |
| 478 DCHECK(did_schedule_); |
| 479 DCHECK(!did_complete_); |
| 480 did_complete_ = true; |
| 481 } |
| 482 |
| 483 bool WorkerPoolTask::HasFinishedRunning() const { |
| 484 return did_run_; |
| 485 } |
| 486 |
| 487 bool WorkerPoolTask::HasCompleted() const { |
| 488 return did_complete_; |
| 489 } |
| 490 |
| 491 GraphNode::GraphNode(internal::WorkerPoolTask* task, unsigned priority) |
| 492 : task_(task), |
| 493 priority_(priority), |
| 494 num_dependencies_(0) { |
| 495 } |
| 496 |
| 497 GraphNode::~GraphNode() { |
| 498 } |
| 499 |
| 500 } // namespace internal |
| 501 |
376 WorkerPool::WorkerPool(size_t num_threads, | 502 WorkerPool::WorkerPool(size_t num_threads, |
377 const std::string& thread_name_prefix) | 503 const std::string& thread_name_prefix) |
378 : in_dispatch_completion_callbacks_(false), | 504 : in_dispatch_completion_callbacks_(false) { |
379 inner_(make_scoped_ptr(new Inner(num_threads, thread_name_prefix))) { | 505 g_task_graph_runner.Pointer()->Register(this); |
380 } | 506 } |
381 | 507 |
382 WorkerPool::~WorkerPool() { | 508 WorkerPool::~WorkerPool() { |
| 509 g_task_graph_runner.Pointer()->Unregister(this); |
383 } | 510 } |
384 | 511 |
385 void WorkerPool::Shutdown() { | 512 void WorkerPool::Shutdown() { |
386 TRACE_EVENT0("cc", "WorkerPool::Shutdown"); | 513 TRACE_EVENT0("cc", "WorkerPool::Shutdown"); |
387 | 514 |
388 DCHECK(!in_dispatch_completion_callbacks_); | 515 DCHECK(!in_dispatch_completion_callbacks_); |
389 | 516 |
390 inner_->Shutdown(); | 517 g_task_graph_runner.Pointer()->WaitForTasksToFinishRunning(this); |
391 } | 518 } |
392 | 519 |
393 void WorkerPool::CheckForCompletedTasks() { | 520 void WorkerPool::CheckForCompletedTasks() { |
394 TRACE_EVENT0("cc", "WorkerPool::CheckForCompletedTasks"); | 521 TRACE_EVENT0("cc", "WorkerPool::CheckForCompletedTasks"); |
395 | 522 |
396 DCHECK(!in_dispatch_completion_callbacks_); | 523 DCHECK(!in_dispatch_completion_callbacks_); |
397 | 524 |
398 TaskVector completed_tasks; | 525 TaskVector completed_tasks; |
399 inner_->CollectCompletedTasks(&completed_tasks); | 526 g_task_graph_runner.Pointer()->CollectCompletedTasks(this, &completed_tasks); |
400 ProcessCompletedTasks(completed_tasks); | 527 ProcessCompletedTasks(completed_tasks); |
401 } | 528 } |
402 | 529 |
403 void WorkerPool::ProcessCompletedTasks( | 530 void WorkerPool::ProcessCompletedTasks( |
404 const TaskVector& completed_tasks) { | 531 const TaskVector& completed_tasks) { |
405 TRACE_EVENT1("cc", "WorkerPool::ProcessCompletedTasks", | 532 TRACE_EVENT1("cc", "WorkerPool::ProcessCompletedTasks", |
406 "completed_task_count", completed_tasks.size()); | 533 "completed_task_count", completed_tasks.size()); |
407 | 534 |
408 // Worker pool instance is not reentrant while processing completed tasks. | 535 // Worker pool instance is not reentrant while processing completed tasks. |
409 in_dispatch_completion_callbacks_ = true; | 536 in_dispatch_completion_callbacks_ = true; |
(...skipping 10 matching lines...) Expand all Loading... |
420 | 547 |
421 in_dispatch_completion_callbacks_ = false; | 548 in_dispatch_completion_callbacks_ = false; |
422 } | 549 } |
423 | 550 |
424 void WorkerPool::SetTaskGraph(TaskGraph* graph) { | 551 void WorkerPool::SetTaskGraph(TaskGraph* graph) { |
425 TRACE_EVENT1("cc", "WorkerPool::SetTaskGraph", | 552 TRACE_EVENT1("cc", "WorkerPool::SetTaskGraph", |
426 "num_tasks", graph->size()); | 553 "num_tasks", graph->size()); |
427 | 554 |
428 DCHECK(!in_dispatch_completion_callbacks_); | 555 DCHECK(!in_dispatch_completion_callbacks_); |
429 | 556 |
430 inner_->SetTaskGraph(graph); | 557 g_task_graph_runner.Pointer()->SetTaskGraph(this, graph); |
431 } | 558 } |
432 | 559 |
433 } // namespace cc | 560 } // namespace cc |
OLD | NEW |