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