OLD | NEW |
1 // Copyright 2016 The Chromium Authors. All rights reserved. | 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 | 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 "base/task_scheduler/scheduler_thread_pool.h" | 5 #include "base/task_scheduler/scheduler_thread_pool.h" |
6 | 6 |
7 #include <utility> | 7 #include <utility> |
8 | 8 |
9 #include "base/bind.h" | 9 #include "base/bind.h" |
10 #include "base/bind_helpers.h" | 10 #include "base/bind_helpers.h" |
11 #include "base/lazy_instance.h" | 11 #include "base/lazy_instance.h" |
12 #include "base/logging.h" | 12 #include "base/logging.h" |
13 #include "base/memory/ptr_util.h" | 13 #include "base/memory/ptr_util.h" |
14 #include "base/sequenced_task_runner.h" | 14 #include "base/sequenced_task_runner.h" |
15 #include "base/task_scheduler/utils.h" | 15 #include "base/task_scheduler/delayed_task_manager.h" |
| 16 #include "base/task_scheduler/task_tracker.h" |
16 #include "base/threading/thread_local.h" | 17 #include "base/threading/thread_local.h" |
17 | 18 |
18 namespace base { | 19 namespace base { |
19 namespace internal { | 20 namespace internal { |
20 | 21 |
21 namespace { | 22 namespace { |
22 | 23 |
23 // SchedulerThreadPool that owns the current thread, if any. | 24 // SchedulerThreadPool that owns the current thread, if any. |
24 LazyInstance<ThreadLocalPointer<const SchedulerTaskExecutor>>::Leaky | 25 LazyInstance<ThreadLocalPointer<const SchedulerThreadPool>>::Leaky |
25 tls_current_thread_pool = LAZY_INSTANCE_INITIALIZER; | 26 tls_current_thread_pool = LAZY_INSTANCE_INITIALIZER; |
26 | 27 |
27 // A task runner that runs tasks with the PARALLEL ExecutionMode. | 28 // A task runner that runs tasks with the PARALLEL ExecutionMode. |
28 class SchedulerParallelTaskRunner : public TaskRunner { | 29 class SchedulerParallelTaskRunner : public TaskRunner { |
29 public: | 30 public: |
30 // Constructs a SchedulerParallelTaskRunner which can be used to post tasks so | 31 // Constructs a SchedulerParallelTaskRunner which can be used to post tasks so |
31 // long as |executor| is alive. | 32 // long as |thread_pool| is alive. |
32 // TODO(robliao): Find a concrete way to manage |executor|'s memory. | 33 // TODO(robliao): Find a concrete way to manage |thread_pool|'s memory. |
33 SchedulerParallelTaskRunner(const TaskTraits& traits, | 34 SchedulerParallelTaskRunner(const TaskTraits& traits, |
34 SchedulerTaskExecutor* executor, | 35 SchedulerThreadPool* thread_pool) |
35 TaskTracker* task_tracker, | 36 : traits_(traits), thread_pool_(thread_pool) {} |
36 DelayedTaskManager* delayed_task_manager) | |
37 : traits_(traits), | |
38 executor_(executor), | |
39 task_tracker_(task_tracker), | |
40 delayed_task_manager_(delayed_task_manager) {} | |
41 | 37 |
42 // TaskRunner: | 38 // TaskRunner: |
43 bool PostDelayedTask(const tracked_objects::Location& from_here, | 39 bool PostDelayedTask(const tracked_objects::Location& from_here, |
44 const Closure& closure, | 40 const Closure& closure, |
45 TimeDelta delay) override { | 41 TimeDelta delay) override { |
46 // Post the task as part of a one-off single-task Sequence. | 42 // Post the task as part of a one-off single-task Sequence. |
47 return PostTaskToExecutor( | 43 return thread_pool_->PostTaskWithSequence( |
48 WrapUnique( | 44 WrapUnique( |
49 new Task(from_here, closure, traits_, | 45 new Task(from_here, closure, traits_, |
50 delay.is_zero() ? TimeTicks() : TimeTicks::Now() + delay)), | 46 delay.is_zero() ? TimeTicks() : TimeTicks::Now() + delay)), |
51 make_scoped_refptr(new Sequence), executor_, task_tracker_, | 47 make_scoped_refptr(new Sequence)); |
52 delayed_task_manager_); | |
53 } | 48 } |
54 | 49 |
55 bool RunsTasksOnCurrentThread() const override { | 50 bool RunsTasksOnCurrentThread() const override { |
56 return tls_current_thread_pool.Get().Get() == executor_; | 51 return tls_current_thread_pool.Get().Get() == thread_pool_; |
57 } | 52 } |
58 | 53 |
59 private: | 54 private: |
60 ~SchedulerParallelTaskRunner() override = default; | 55 ~SchedulerParallelTaskRunner() override = default; |
61 | 56 |
62 const TaskTraits traits_; | 57 const TaskTraits traits_; |
63 SchedulerTaskExecutor* const executor_; | 58 SchedulerThreadPool* const thread_pool_; |
64 TaskTracker* const task_tracker_; | |
65 DelayedTaskManager* const delayed_task_manager_; | |
66 | 59 |
67 DISALLOW_COPY_AND_ASSIGN(SchedulerParallelTaskRunner); | 60 DISALLOW_COPY_AND_ASSIGN(SchedulerParallelTaskRunner); |
68 }; | 61 }; |
69 | 62 |
70 // A task runner that runs tasks with the SEQUENCED ExecutionMode. | 63 // A task runner that runs tasks with the SEQUENCED ExecutionMode. |
71 class SchedulerSequencedTaskRunner : public SequencedTaskRunner { | 64 class SchedulerSequencedTaskRunner : public SequencedTaskRunner { |
72 public: | 65 public: |
73 // Constructs a SchedulerParallelTaskRunner which can be used to post tasks so | 66 // Constructs a SchedulerSequencedTaskRunner which can be used to post tasks |
74 // long as |executor| is alive. | 67 // so long as |thread_pool| is alive. |
75 // TODO(robliao): Find a concrete way to manage |executor|'s memory. | 68 // TODO(robliao): Find a concrete way to manage |thread_pool|'s memory. |
76 SchedulerSequencedTaskRunner(const TaskTraits& traits, | 69 SchedulerSequencedTaskRunner(const TaskTraits& traits, |
77 SchedulerTaskExecutor* executor, | 70 SchedulerThreadPool* thread_pool) |
78 TaskTracker* task_tracker, | 71 : traits_(traits), thread_pool_(thread_pool) {} |
79 DelayedTaskManager* delayed_task_manager) | |
80 : traits_(traits), | |
81 executor_(executor), | |
82 task_tracker_(task_tracker), | |
83 delayed_task_manager_(delayed_task_manager) {} | |
84 | 72 |
85 // SequencedTaskRunner: | 73 // SequencedTaskRunner: |
86 bool PostDelayedTask(const tracked_objects::Location& from_here, | 74 bool PostDelayedTask(const tracked_objects::Location& from_here, |
87 const Closure& closure, | 75 const Closure& closure, |
88 TimeDelta delay) override { | 76 TimeDelta delay) override { |
89 // Post the task as part of |sequence|. | 77 // Post the task as part of |sequence|. |
90 return PostTaskToExecutor( | 78 return thread_pool_->PostTaskWithSequence( |
91 WrapUnique( | 79 WrapUnique( |
92 new Task(from_here, closure, traits_, | 80 new Task(from_here, closure, traits_, |
93 delay.is_zero() ? TimeTicks() : TimeTicks::Now() + delay)), | 81 delay.is_zero() ? TimeTicks() : TimeTicks::Now() + delay)), |
94 sequence_, executor_, task_tracker_, delayed_task_manager_); | 82 sequence_); |
95 } | 83 } |
96 | 84 |
97 bool PostNonNestableDelayedTask(const tracked_objects::Location& from_here, | 85 bool PostNonNestableDelayedTask(const tracked_objects::Location& from_here, |
98 const Closure& closure, | 86 const Closure& closure, |
99 base::TimeDelta delay) override { | 87 base::TimeDelta delay) override { |
100 // Tasks are never nested within the task scheduler. | 88 // Tasks are never nested within the task scheduler. |
101 return PostDelayedTask(from_here, closure, delay); | 89 return PostDelayedTask(from_here, closure, delay); |
102 } | 90 } |
103 | 91 |
104 bool RunsTasksOnCurrentThread() const override { | 92 bool RunsTasksOnCurrentThread() const override { |
105 return tls_current_thread_pool.Get().Get() == executor_; | 93 return tls_current_thread_pool.Get().Get() == thread_pool_; |
106 } | 94 } |
107 | 95 |
108 private: | 96 private: |
109 ~SchedulerSequencedTaskRunner() override = default; | 97 ~SchedulerSequencedTaskRunner() override = default; |
110 | 98 |
111 // Sequence for all Tasks posted through this TaskRunner. | 99 // Sequence for all Tasks posted through this TaskRunner. |
112 const scoped_refptr<Sequence> sequence_ = new Sequence; | 100 const scoped_refptr<Sequence> sequence_ = new Sequence; |
113 | 101 |
114 const TaskTraits traits_; | 102 const TaskTraits traits_; |
115 SchedulerTaskExecutor* const executor_; | 103 SchedulerThreadPool* const thread_pool_; |
116 TaskTracker* const task_tracker_; | |
117 DelayedTaskManager* const delayed_task_manager_; | |
118 | 104 |
119 DISALLOW_COPY_AND_ASSIGN(SchedulerSequencedTaskRunner); | 105 DISALLOW_COPY_AND_ASSIGN(SchedulerSequencedTaskRunner); |
120 }; | 106 }; |
121 | 107 |
122 } // namespace | 108 } // namespace |
123 | 109 |
124 class SchedulerThreadPool::SchedulerWorkerThreadDelegateImpl | 110 class SchedulerThreadPool::SchedulerWorkerThreadDelegateImpl |
125 : public SchedulerWorkerThread::Delegate { | 111 : public SchedulerWorkerThread::Delegate { |
126 public: | 112 public: |
127 SchedulerWorkerThreadDelegateImpl( | 113 SchedulerWorkerThreadDelegateImpl( |
(...skipping 31 matching lines...) Expand 10 before | Expand all | Expand 10 after Loading... |
159 if (thread_pool->Initialize(thread_priority, max_threads)) | 145 if (thread_pool->Initialize(thread_priority, max_threads)) |
160 return thread_pool; | 146 return thread_pool; |
161 return nullptr; | 147 return nullptr; |
162 } | 148 } |
163 | 149 |
164 scoped_refptr<TaskRunner> SchedulerThreadPool::CreateTaskRunnerWithTraits( | 150 scoped_refptr<TaskRunner> SchedulerThreadPool::CreateTaskRunnerWithTraits( |
165 const TaskTraits& traits, | 151 const TaskTraits& traits, |
166 ExecutionMode execution_mode) { | 152 ExecutionMode execution_mode) { |
167 switch (execution_mode) { | 153 switch (execution_mode) { |
168 case ExecutionMode::PARALLEL: | 154 case ExecutionMode::PARALLEL: |
169 return make_scoped_refptr(new SchedulerParallelTaskRunner( | 155 return make_scoped_refptr(new SchedulerParallelTaskRunner(traits, this)); |
170 traits, this, task_tracker_, delayed_task_manager_)); | |
171 | 156 |
172 case ExecutionMode::SEQUENCED: | 157 case ExecutionMode::SEQUENCED: |
173 return make_scoped_refptr(new SchedulerSequencedTaskRunner( | 158 return make_scoped_refptr(new SchedulerSequencedTaskRunner(traits, this)); |
174 traits, this, task_tracker_, delayed_task_manager_)); | |
175 | 159 |
176 case ExecutionMode::SINGLE_THREADED: | 160 case ExecutionMode::SINGLE_THREADED: |
177 // TODO(fdoray): Support SINGLE_THREADED TaskRunners. | 161 // TODO(fdoray): Support SINGLE_THREADED TaskRunners. |
178 NOTREACHED(); | 162 NOTREACHED(); |
179 return nullptr; | 163 return nullptr; |
180 } | 164 } |
181 | 165 |
182 NOTREACHED(); | 166 NOTREACHED(); |
183 return nullptr; | 167 return nullptr; |
184 } | 168 } |
(...skipping 24 matching lines...) Expand all Loading... |
209 } | 193 } |
210 | 194 |
211 void SchedulerThreadPool::JoinForTesting() { | 195 void SchedulerThreadPool::JoinForTesting() { |
212 for (const auto& worker_thread : worker_threads_) | 196 for (const auto& worker_thread : worker_threads_) |
213 worker_thread->JoinForTesting(); | 197 worker_thread->JoinForTesting(); |
214 | 198 |
215 DCHECK(!join_for_testing_returned_.IsSignaled()); | 199 DCHECK(!join_for_testing_returned_.IsSignaled()); |
216 join_for_testing_returned_.Signal(); | 200 join_for_testing_returned_.Signal(); |
217 } | 201 } |
218 | 202 |
219 void SchedulerThreadPool::PostTaskWithSequence( | 203 bool SchedulerThreadPool::PostTaskWithSequence( |
220 std::unique_ptr<Task> task, | 204 std::unique_ptr<Task> task, |
221 scoped_refptr<Sequence> sequence) { | 205 scoped_refptr<Sequence> sequence) { |
222 DCHECK(task); | 206 DCHECK(task); |
223 DCHECK(sequence); | 207 DCHECK(sequence); |
224 | 208 |
225 const bool sequence_was_empty = AddTaskToSequenceAndPriorityQueue( | 209 if (!task_tracker_->WillPostTask(task.get())) |
226 std::move(task), std::move(sequence), &shared_priority_queue_); | 210 return false; |
227 | 211 |
228 // No thread has already been woken up to run Tasks from |sequence| if it was | 212 if (task->delayed_run_time.is_null()) { |
229 // empty before |task| was inserted into it. | 213 PostTaskWithSequenceNow(std::move(task), std::move(sequence)); |
230 if (sequence_was_empty) | 214 } else { |
| 215 delayed_task_manager_->AddDelayedTask(std::move(task), std::move(sequence), |
| 216 this); |
| 217 } |
| 218 |
| 219 return true; |
| 220 } |
| 221 |
| 222 void SchedulerThreadPool::PostTaskWithSequenceNow( |
| 223 std::unique_ptr<Task> task, |
| 224 scoped_refptr<Sequence> sequence) { |
| 225 DCHECK(task); |
| 226 DCHECK(sequence); |
| 227 |
| 228 // Confirm that |task| is ready to run (its delayed run time is either null or |
| 229 // in the past). |
| 230 DCHECK_LE(task->delayed_run_time, TimeTicks::Now()); |
| 231 |
| 232 const bool sequence_was_empty = sequence->PushTask(std::move(task)); |
| 233 if (sequence_was_empty) { |
| 234 // Insert |sequence| in |shared_priority_queue_| if it was empty before |
| 235 // |task| was inserted into it. Otherwise, one of these must be true: |
| 236 // - |sequence| is already in a PriorityQueue (not necessarily |
| 237 // |shared_priority_queue_|), or, |
| 238 // - A worker thread is running a Task from |sequence|. It will insert |
| 239 // |sequence| in a PriorityQueue once it's done running the Task. |
| 240 const auto sequence_sort_key = sequence->GetSortKey(); |
| 241 shared_priority_queue_.BeginTransaction()->Push( |
| 242 WrapUnique(new PriorityQueue::SequenceAndSortKey(std::move(sequence), |
| 243 sequence_sort_key))); |
| 244 |
| 245 // Wake up a worker thread to process |sequence|. |
231 WakeUpOneThread(); | 246 WakeUpOneThread(); |
| 247 } |
232 } | 248 } |
233 | 249 |
234 SchedulerThreadPool::SchedulerWorkerThreadDelegateImpl:: | 250 SchedulerThreadPool::SchedulerWorkerThreadDelegateImpl:: |
235 SchedulerWorkerThreadDelegateImpl( | 251 SchedulerWorkerThreadDelegateImpl( |
236 SchedulerThreadPool* outer, | 252 SchedulerThreadPool* outer, |
237 const EnqueueSequenceCallback& enqueue_sequence_callback) | 253 const EnqueueSequenceCallback& enqueue_sequence_callback) |
238 : outer_(outer), enqueue_sequence_callback_(enqueue_sequence_callback) {} | 254 : outer_(outer), enqueue_sequence_callback_(enqueue_sequence_callback) {} |
239 | 255 |
240 SchedulerThreadPool::SchedulerWorkerThreadDelegateImpl:: | 256 SchedulerThreadPool::SchedulerWorkerThreadDelegateImpl:: |
241 ~SchedulerWorkerThreadDelegateImpl() = default; | 257 ~SchedulerWorkerThreadDelegateImpl() = default; |
(...skipping 93 matching lines...) Expand 10 before | Expand all | Expand 10 after Loading... |
335 AutoSchedulerLock auto_lock(idle_worker_threads_stack_lock_); | 351 AutoSchedulerLock auto_lock(idle_worker_threads_stack_lock_); |
336 | 352 |
337 if (idle_worker_threads_stack_.Empty()) | 353 if (idle_worker_threads_stack_.Empty()) |
338 return nullptr; | 354 return nullptr; |
339 | 355 |
340 return idle_worker_threads_stack_.Pop(); | 356 return idle_worker_threads_stack_.Pop(); |
341 } | 357 } |
342 | 358 |
343 } // namespace internal | 359 } // namespace internal |
344 } // namespace base | 360 } // namespace base |
OLD | NEW |