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

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

Powered by Google App Engine
This is Rietveld 408576698