Chromium Code Reviews
chromiumcodereview-hr@appspot.gserviceaccount.com (chromiumcodereview-hr) | Please choose your nickname with Settings | Help | Chromium Project | Gerrit Changes | Sign out
(127)

Side by Side Diff: cc/resources/worker_pool.cc

Issue 73923003: Shared Raster Worker Threads (Closed) Base URL: https://chromium.googlesource.com/chromium/src.git@master
Patch Set: WIP - make API to retrieve number of raster threads consistent + nits Created 6 years, 12 months ago
Use n/p to move between diff chunks; N/P to move between comments. Draft comments are only viewable by you.
Jump to:
View unified diff | Download patch
OLDNEW
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
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
OLDNEW

Powered by Google App Engine
This is Rietveld 408576698