OLD | NEW |
---|---|
(Empty) | |
1 // Copyright 2016 The Chromium Authors. All rights reserved. | |
2 // Use of this source code is governed by a BSD-style license that can be | |
3 // found in the LICENSE file. | |
4 | |
5 #include "base/task_scheduler/worker_thread.h" | |
6 | |
7 #include <utility> | |
8 | |
9 #include "base/bind.h" | |
10 #include "base/debug/task_annotator.h" | |
11 #include "base/logging.h" | |
12 #include "base/task_scheduler/delayed_task_manager.h" | |
13 #include "base/task_scheduler/priority_queue.h" | |
14 #include "base/task_scheduler/shutdown_manager.h" | |
15 #include "base/task_scheduler/utils.h" | |
16 #include "base/time/time.h" | |
17 #include "build/build_config.h" | |
18 | |
19 namespace base { | |
20 namespace task_scheduler { | |
21 | |
22 namespace { | |
23 | |
24 // A task runner that runs tasks on a single WorkerThread. | |
25 class SchedulerSingleThreadTaskRunner : public SingleThreadTaskRunner { | |
26 public: | |
27 // Tasks posted through this task runner have |traits| and are inserted in | |
28 // |single_thread_priority_queue|. |delayed_task_manager| is used to post | |
29 // delayed tasks. |shutdown_manager| is notified when a task is posted. | |
30 SchedulerSingleThreadTaskRunner(const TaskTraits& traits, | |
31 PriorityQueue* single_thread_priority_queue, | |
32 DelayedTaskManager* delayed_task_manager, | |
33 ShutdownManager* shutdown_manager); | |
34 | |
35 // SingleThreadTaskRunner: | |
36 bool PostNonNestableDelayedTask(const tracked_objects::Location& from_here, | |
37 const Closure& task, | |
38 TimeDelta delay) override; | |
39 bool PostDelayedTask(const tracked_objects::Location& from_here, | |
40 const Closure& closure, | |
41 TimeDelta delay) override; | |
42 bool RunsTasksOnCurrentThread() const override; | |
43 | |
44 private: | |
45 ~SchedulerSingleThreadTaskRunner() override; | |
46 | |
47 TaskTraits traits_; | |
48 scoped_refptr<Sequence> sequence_; | |
49 PriorityQueue* priority_queue_; | |
50 DelayedTaskManager* delayed_task_manager_; | |
51 ShutdownManager* shutdown_manager_; | |
52 | |
53 DISALLOW_COPY_AND_ASSIGN(SchedulerSingleThreadTaskRunner); | |
54 }; | |
55 | |
56 SchedulerSingleThreadTaskRunner::SchedulerSingleThreadTaskRunner( | |
57 const TaskTraits& traits, | |
58 PriorityQueue* priority_queue, | |
59 DelayedTaskManager* delayed_task_manager, | |
60 ShutdownManager* shutdown_manager) | |
61 : traits_(traits), | |
62 sequence_(new Sequence), | |
63 priority_queue_(priority_queue), | |
64 delayed_task_manager_(delayed_task_manager), | |
65 shutdown_manager_(shutdown_manager) {} | |
66 | |
67 bool SchedulerSingleThreadTaskRunner::PostDelayedTask( | |
68 const tracked_objects::Location& from_here, | |
69 const Closure& closure, | |
70 TimeDelta delay) { | |
71 Task task(from_here, closure, traits_, TimeTicks::Now()); | |
72 if (!delay.is_zero()) | |
73 task.delayed_run_time = task.post_time + delay; | |
74 PostTaskHelper(task, sequence_, priority_queue_, shutdown_manager_, | |
75 delayed_task_manager_); | |
76 return true; | |
77 } | |
78 | |
79 bool SchedulerSingleThreadTaskRunner::RunsTasksOnCurrentThread() const { | |
80 // TODO(fdoray): Return true only if tasks posted may actually run on the | |
81 // current thread. It is valid, but not ideal, to always return true. | |
82 return true; | |
83 } | |
84 | |
85 bool SchedulerSingleThreadTaskRunner::PostNonNestableDelayedTask( | |
86 const tracked_objects::Location& from_here, | |
87 const Closure& task, | |
88 TimeDelta delay) { | |
89 return PostDelayedTask(from_here, task, delay); | |
90 } | |
91 | |
92 SchedulerSingleThreadTaskRunner::~SchedulerSingleThreadTaskRunner() = default; | |
93 | |
94 } // namespace | |
95 | |
96 scoped_ptr<WorkerThread> WorkerThread::CreateWorkerThread( | |
97 ThreadPriority thread_priority, | |
98 PriorityQueue* shared_priority_queue, | |
99 const ReinsertSequenceCallback& reinsert_sequence_callback, | |
100 const BecomesIdleCallback& becomes_idle_callback, | |
101 DelayedTaskManager* delayed_task_manager, | |
102 ShutdownManager* shutdown_manager) { | |
103 scoped_ptr<WorkerThread> worker_thread(new WorkerThread( | |
104 thread_priority, shared_priority_queue, reinsert_sequence_callback, | |
105 becomes_idle_callback, delayed_task_manager, shutdown_manager)); | |
106 return worker_thread->IsValid() ? std::move(worker_thread) | |
107 : scoped_ptr<WorkerThread>(); | |
108 } | |
109 | |
110 WorkerThread::~WorkerThread() = default; | |
111 | |
112 void WorkerThread::WakeUp() { | |
113 wakeup_event_.Signal(); | |
114 } | |
115 | |
116 scoped_refptr<SingleThreadTaskRunner> WorkerThread::CreateTaskRunnerWithTraits( | |
117 const TaskTraits& traits, | |
118 ExecutionMode execution_mode) { | |
119 #if defined(OS_WIN) | |
120 DCHECK(execution_mode == ExecutionMode::SINGLE_THREADED || | |
121 execution_mode == ExecutionMode::SINGLE_THREADED_COM_STA); | |
122 #else | |
123 DCHECK(execution_mode == ExecutionMode::SINGLE_THREADED); | |
124 #endif // defined(OS_WIN) | |
125 | |
126 return scoped_refptr<SingleThreadTaskRunner>( | |
127 new SchedulerSingleThreadTaskRunner( | |
128 traits, &single_thread_priority_queue_, delayed_task_manager_, | |
129 shutdown_manager_)); | |
130 } | |
131 | |
132 bool WorkerThread::HasSingleThreadedTasks() const { | |
133 return !single_thread_priority_queue_.UnsynchronizedEmpty() || | |
134 is_running_single_threaded_task_; | |
135 } | |
136 | |
137 void WorkerThread::JoinForTesting() { | |
138 DCHECK(shutdown_manager_->shutdown_completed()); | |
139 WakeUp(); | |
140 PlatformThread::Join(thread_handle_); | |
141 } | |
142 | |
143 WorkerThread::WorkerThread( | |
144 ThreadPriority thread_priority, | |
145 PriorityQueue* shared_priority_queue, | |
146 const ReinsertSequenceCallback& reinsert_sequence_callback, | |
147 const BecomesIdleCallback& becomes_idle_callback, | |
148 DelayedTaskManager* delayed_task_manager, | |
149 ShutdownManager* shutdown_manager) | |
150 : wakeup_event_(false, false), | |
151 is_running_single_threaded_task_(false), | |
152 single_thread_priority_queue_( | |
153 Bind(&WorkerThread::WakeUp, Unretained(this)), | |
154 shared_priority_queue), | |
155 shared_priority_queue_(shared_priority_queue), | |
156 reinsert_sequence_callback_(reinsert_sequence_callback), | |
157 becomes_idle_callback_(becomes_idle_callback), | |
158 delayed_task_manager_(delayed_task_manager), | |
159 shutdown_manager_(shutdown_manager) { | |
160 DCHECK(shared_priority_queue_); | |
161 DCHECK(!reinsert_sequence_callback.is_null()); | |
162 DCHECK(!becomes_idle_callback.is_null()); | |
163 DCHECK(delayed_task_manager_); | |
164 DCHECK(shutdown_manager_); | |
165 | |
166 #if defined(OS_MACOSX) | |
167 // Mac only supports 2 priorities. crbug.com/554651 | |
168 if (thread_priority != ThreadPriority::NORMAL && | |
169 thread_priority != ThreadPriority::REALTIME_AUDIO) { | |
170 thread_priority = ThreadPriority::NORMAL; | |
171 } | |
172 #endif // defined(OS_MACOSX) | |
173 | |
174 const size_t kDefaultStackSize = 0; | |
175 PlatformThread::CreateWithPriority(kDefaultStackSize, this, &thread_handle_, | |
176 thread_priority); | |
177 } | |
178 | |
179 bool WorkerThread::IsValid() const { | |
180 return !thread_handle_.is_null(); | |
181 } | |
182 | |
183 scoped_refptr<Sequence> WorkerThread::GetWork() { | |
184 scoped_ptr<PriorityQueue::Transaction> shared_transaction( | |
185 shared_priority_queue_->BeginTransaction()); | |
186 SequenceSortKey shared_sort_key; | |
187 scoped_refptr<Sequence> shared_sequence = | |
188 shared_transaction->PeekSequence(&shared_sort_key); | |
189 | |
190 scoped_ptr<PriorityQueue::Transaction> single_thread_transaction( | |
191 single_thread_priority_queue_.BeginTransaction()); | |
192 SequenceSortKey single_thread_sort_key; | |
193 scoped_refptr<Sequence> single_thread_sequence = | |
194 single_thread_transaction->PeekSequence(&single_thread_sort_key); | |
195 | |
196 if (single_thread_sequence.get() == nullptr && | |
197 shared_sequence.get() == nullptr) { | |
198 return scoped_refptr<Sequence>(); | |
199 } | |
200 | |
201 if (single_thread_sequence.get() == nullptr || | |
202 (shared_sequence.get() != nullptr && | |
203 single_thread_sort_key < shared_sort_key)) { | |
204 shared_transaction->PopSequence(); | |
205 return shared_sequence; | |
206 } | |
207 | |
208 DCHECK(single_thread_sequence.get()); | |
209 | |
210 is_running_single_threaded_task_ = true; | |
211 single_thread_transaction->PopSequence(); | |
212 return single_thread_sequence; | |
213 } | |
214 | |
215 void WorkerThread::ReinsertSequenceInSingleThreadPriorityQueue( | |
216 scoped_refptr<Sequence> sequence) { | |
217 // Get the sort key of |sequence| before creating a priority queue | |
218 // transaction, to avoid holding 2 locks at the same time. | |
219 SequenceSortKey sort_key = sequence->GetSortKey(); | |
220 | |
221 // Insert the sequence in the single-thread priority queue. | |
222 single_thread_priority_queue_.BeginTransaction()->PushSequence(sequence, | |
223 sort_key); | |
224 } | |
225 | |
226 void WorkerThread::WaitUntilWorkIsAvailable() { | |
227 const TimeTicks next_delayed_task_ready_time = | |
228 delayed_task_manager_->GetNextDelayedTaskReadyTime(); | |
229 | |
230 if (next_delayed_task_ready_time.is_null()) { | |
231 // There is no delayed tasks. Wait until |wakeup_event_| is signaled. | |
232 wakeup_event_.Wait(); | |
233 } else { | |
234 // There is delayed tasks. Wait until either a delayed task becomes ready | |
235 // for execution or |wakeup_event_| is signaled. Note: Multiple threads | |
236 // sharing the same DelayedTaskManager may wake up at the same time when a | |
237 // delayed task becomes ready for execution. This isn't optimal. However, | |
238 // since most delayed tasks should be posted to BACKGROUND thread pools | |
239 // (which have a single thread), this behavior shouldn't occur frequently. | |
240 const TimeDelta wait_time = next_delayed_task_ready_time - TimeTicks::Now(); | |
241 if (wait_time.InMilliseconds() > 0) | |
242 wakeup_event_.TimedWait(wait_time); | |
243 } | |
244 } | |
245 | |
246 void WorkerThread::ThreadMain() { | |
247 for (;;) { | |
robliao
2016/02/11 21:56:27
Maybe this should be while(!shutdown_manager->shut
fdoray
2016/02/12 04:16:20
Done.
| |
248 // Get the sequence containing the next task to execute. | |
249 scoped_refptr<Sequence> sequence = GetWork(); | |
250 if (sequence.get() == nullptr) { | |
251 // Add the thread to the stack of idle threads of the parent thread pool. | |
252 becomes_idle_callback_.Run(this); | |
253 | |
254 // Check one more time whether there is pending work. Without this, it | |
255 // could be that work has been added to |shared_priority_queue_| after the | |
256 // first call to GetWork() and before this thread was added to the stack | |
257 // of idle threads. In such a case, |wake_up_event_| hasn't been signaled | |
258 // because this thread wasn't in the stack of idle threads. However, this | |
259 // thread is needed to execute the newly added work. | |
260 sequence = GetWork(); | |
robliao
2016/02/11 22:30:07
Can we remove this somehow?
fdoray
2016/02/12 04:16:20
We could if we built something that atomically che
| |
261 | |
262 if (sequence.get() == nullptr) { | |
263 WaitUntilWorkIsAvailable(); | |
264 sequence = GetWork(); | |
265 } | |
266 } | |
267 | |
268 if (sequence.get() != nullptr) { | |
269 // Peek the next task in the sequence. | |
270 const Task* task = sequence->PeekTask(); | |
271 DCHECK(task); | |
robliao
2016/02/11 22:30:07
Remove DCHECK.
fdoray
2016/02/12 04:16:20
Done.
| |
272 const TaskShutdownBehavior shutdown_behavior = | |
273 task->traits.shutdown_behavior(); | |
274 | |
275 // Run the task. | |
276 if (shutdown_manager_->ShouldScheduleTask(shutdown_behavior)) { | |
277 debug::TaskAnnotator task_annotator; | |
278 task_annotator.RunTask("task_scheduler::PostTask", *task); | |
279 shutdown_manager_->DidExecuteTask(shutdown_behavior); | |
280 } | |
281 | |
282 // Pop the task from the sequence. | |
283 size_t new_num_tasks_in_sequence; | |
284 sequence->PopTask(&new_num_tasks_in_sequence); | |
285 | |
286 // Put the sequence back in the appropriate priority queue. | |
287 if (new_num_tasks_in_sequence > 0) { | |
288 if (is_running_single_threaded_task_) | |
289 ReinsertSequenceInSingleThreadPriorityQueue(sequence); | |
290 else | |
291 reinsert_sequence_callback_.Run(sequence, this); | |
292 } | |
293 | |
294 // Note that the thread is no longer running a single-threaded task. | |
295 is_running_single_threaded_task_ = false; | |
296 } | |
297 | |
298 // Exit the thread if shutdown is complete. | |
299 if (shutdown_manager_->shutdown_completed()) | |
300 break; | |
301 | |
302 // Post delayed tasks that are ready for execution. | |
303 delayed_task_manager_->PostReadyTasks(); | |
304 } | |
305 } | |
306 | |
307 } // namespace task_scheduler | |
308 } // namespace base | |
OLD | NEW |