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