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/scheduler_thread_pool.h" | |
6 | |
7 #include <utility> | |
8 | |
9 #include "base/bind.h" | |
10 #include "base/bind_helpers.h" | |
11 #include "base/lazy_instance.h" | |
12 #include "base/logging.h" | |
13 #include "base/memory/ptr_util.h" | |
14 #include "base/task_scheduler/utils.h" | |
15 #include "base/threading/thread_local.h" | |
16 | |
17 namespace base { | |
18 namespace internal { | |
19 | |
20 namespace { | |
21 | |
22 // SchedulerThreadPool that owns the current thread, if any. | |
23 LazyInstance<ThreadLocalPointer<const SchedulerTaskExecutor>>::Leaky | |
24 tls_current_thread_pool = LAZY_INSTANCE_INITIALIZER; | |
25 | |
26 // A task runner that runs tasks with the PARALLEL ExecutionMode. | |
27 class SchedulerParallelTaskRunner : public TaskRunner { | |
28 public: | |
29 SchedulerParallelTaskRunner(const TaskTraits& traits, | |
30 TaskTracker* task_tracker, | |
31 SchedulerTaskExecutor* executor) | |
32 : traits_(traits), task_tracker_(task_tracker), executor_(executor) {} | |
33 | |
34 // TaskRunner: | |
35 bool PostDelayedTask(const tracked_objects::Location& from_here, | |
36 const Closure& closure, | |
37 TimeDelta delay) override { | |
38 // Post the task as part of a one-off single-task Sequence. | |
39 return PostTaskHelper(from_here, closure, traits_, delay, | |
40 make_scoped_refptr(new Sequence), executor_, | |
41 task_tracker_); | |
42 } | |
43 | |
44 bool RunsTasksOnCurrentThread() const override { | |
45 return tls_current_thread_pool.Get().Get() == executor_; | |
46 } | |
47 | |
48 private: | |
49 ~SchedulerParallelTaskRunner() override = default; | |
50 | |
51 const TaskTraits traits_; | |
52 TaskTracker* const task_tracker_; | |
53 | |
54 // TODO(robliao): Use a WeakPtr instead of a raw pointer. A | |
55 // SchedulerTaskExecutor is never deleted in production, but it's better not | |
56 // to spread this assumption throughout the scheduler. | |
gab
2016/04/14 19:34:16
A WeakPtr isn't ever possible here as PostTask is
fdoray
2016/04/14 23:46:58
Done.
| |
57 SchedulerTaskExecutor* const executor_; | |
58 | |
59 DISALLOW_COPY_AND_ASSIGN(SchedulerParallelTaskRunner); | |
60 }; | |
61 | |
62 } // namespace | |
63 | |
64 class SchedulerThreadPool::SchedulerWorkerThreadDelegateImpl | |
65 : public SchedulerWorkerThread::Delegate { | |
66 public: | |
67 SchedulerWorkerThreadDelegateImpl( | |
68 SchedulerThreadPool* outer, | |
69 const EnqueueSequenceCallback enqueue_sequence_callback); | |
70 ~SchedulerWorkerThreadDelegateImpl() override; | |
71 | |
72 // SchedulerWorkerThread::Delegate: | |
73 void OnMainEntry() override; | |
74 scoped_refptr<Sequence> GetWork( | |
75 SchedulerWorkerThread* worker_thread) override; | |
76 void EnqueueSequence(scoped_refptr<Sequence> sequence) override; | |
77 | |
78 private: | |
79 SchedulerThreadPool* outer_; | |
80 const EnqueueSequenceCallback enqueue_sequence_callback_; | |
81 | |
82 DISALLOW_COPY_AND_ASSIGN(SchedulerWorkerThreadDelegateImpl); | |
83 }; | |
84 | |
85 SchedulerThreadPool::~SchedulerThreadPool() { | |
86 // SchedulerThreadPool should never be deleted in production unless its | |
87 // initialization failed. | |
88 DCHECK(join_for_testing_returned_.IsSignaled() || worker_threads_.empty()); | |
89 } | |
90 | |
91 std::unique_ptr<SchedulerThreadPool> SchedulerThreadPool::CreateThreadPool( | |
92 ThreadPriority thread_priority, | |
93 size_t max_threads, | |
94 const EnqueueSequenceCallback& enqueue_sequence_callback, | |
95 TaskTracker* task_tracker) { | |
96 std::unique_ptr<SchedulerThreadPool> thread_pool( | |
97 new SchedulerThreadPool(enqueue_sequence_callback, task_tracker)); | |
98 if (thread_pool->Initialize(thread_priority, max_threads)) | |
99 return thread_pool; | |
100 return nullptr; | |
101 } | |
102 | |
103 scoped_refptr<TaskRunner> SchedulerThreadPool::CreateTaskRunnerWithTraits( | |
104 const TaskTraits& traits, | |
105 ExecutionMode execution_mode) { | |
106 switch (execution_mode) { | |
107 case ExecutionMode::PARALLEL: | |
108 return make_scoped_refptr( | |
109 new SchedulerParallelTaskRunner(traits, task_tracker_, this)); | |
110 | |
111 case ExecutionMode::SEQUENCED: | |
112 case ExecutionMode::SINGLE_THREADED: | |
113 // TODO(fdoray): Support SEQUENCED and SINGLE_THREADED TaskRunners. | |
114 NOTREACHED(); | |
115 return nullptr; | |
116 } | |
117 | |
118 NOTREACHED(); | |
119 return nullptr; | |
120 } | |
121 | |
122 void SchedulerThreadPool::EnqueueSequence( | |
123 scoped_refptr<Sequence> sequence, | |
124 const SequenceSortKey& sequence_sort_key) { | |
125 shared_priority_queue_.BeginTransaction()->Push( | |
126 WrapUnique(new PriorityQueue::SequenceAndSortKey(std::move(sequence), | |
127 sequence_sort_key))); | |
128 | |
129 // The thread calling this method just ran a Task from |sequence| and will | |
130 // soon try to get another Sequence from which to run a Task. If the thread | |
131 // belongs to this pool, it will get that Sequence from | |
132 // |shared_priority_queue_|. When that's the case, there is no need to wake up | |
133 // another thread after |sequence| is inserted in |shared_priority_queue_|. If | |
134 // we did wake up another thread, we would waste resources by having more | |
135 // threads trying to get a Sequence from |shared_priority_queue_| than the | |
136 // number of Sequences in it. | |
137 if (tls_current_thread_pool.Get().Get() != this) | |
138 WakeUpOneThread(); | |
139 } | |
140 | |
141 void SchedulerThreadPool::WaitForAllWorkerThreadsIdleForTesting() { | |
142 AutoSchedulerLock auto_lock(idle_worker_threads_stack_lock_); | |
143 while (idle_worker_threads_stack_.size() < worker_threads_.size()) | |
144 idle_worker_threads_stack_cv_for_testing_->Wait(); | |
145 } | |
146 | |
147 void SchedulerThreadPool::JoinForTesting() { | |
148 for (const auto& worker_thread : worker_threads_) | |
149 worker_thread->JoinForTesting(); | |
150 | |
151 DCHECK(!join_for_testing_returned_.IsSignaled()); | |
152 join_for_testing_returned_.Signal(); | |
153 } | |
154 | |
155 void SchedulerThreadPool::PostTaskWithSequence( | |
156 std::unique_ptr<Task> task, | |
157 scoped_refptr<Sequence> sequence) { | |
158 DCHECK(task); | |
159 DCHECK(sequence); | |
160 | |
161 // Verify that we are passed |task|'s delayed run time (i.e. |task| is ready | |
gab
2016/04/14 19:34:16
s/passed/past/
rm "we", e.g.:
// Confirm that |t
fdoray
2016/04/14 23:46:58
Done (removed the DCHECK here, kept in in utils.cc
| |
162 // to run). | |
163 DCHECK_LE(task->delayed_run_time, TimeTicks::Now()); | |
164 | |
165 const bool sequence_was_empty = PostTaskWithSequenceHelper( | |
166 std::move(task), std::move(sequence), &shared_priority_queue_); | |
167 | |
168 // No thread has already been woken up to run Tasks from |sequence| if it was | |
169 // empty before |task| was inserted into it. | |
170 if (sequence_was_empty) | |
171 WakeUpOneThread(); | |
172 } | |
173 | |
174 SchedulerThreadPool::SchedulerWorkerThreadDelegateImpl:: | |
175 SchedulerWorkerThreadDelegateImpl( | |
176 SchedulerThreadPool* outer, | |
177 const EnqueueSequenceCallback enqueue_sequence_callback) | |
178 : outer_(outer), enqueue_sequence_callback_(enqueue_sequence_callback) {} | |
179 | |
180 SchedulerThreadPool::SchedulerWorkerThreadDelegateImpl:: | |
181 ~SchedulerWorkerThreadDelegateImpl() = default; | |
182 | |
183 void SchedulerThreadPool::SchedulerWorkerThreadDelegateImpl::OnMainEntry() { | |
184 DCHECK(!tls_current_thread_pool.Get().Get()); | |
185 tls_current_thread_pool.Get().Set(outer_); | |
186 } | |
187 | |
188 scoped_refptr<Sequence> | |
189 SchedulerThreadPool::SchedulerWorkerThreadDelegateImpl::GetWork( | |
190 SchedulerWorkerThread* worker_thread) { | |
191 std::unique_ptr<PriorityQueue::Transaction> transaction( | |
192 outer_->shared_priority_queue_.BeginTransaction()); | |
193 const auto sequence_and_sort_key = transaction->Peek(); | |
194 | |
195 if (sequence_and_sort_key.is_null()) { | |
196 // |transaction| is kept alive while |worker_thread| is added to | |
197 // |idle_worker_threads_stack_| to avoid this race: | |
198 // 1. This thread creates a Transaction, finds |shared_priority_queue_| | |
199 // empty and ends the Transaction. | |
200 // 2. Other thread creates a Transaction, inserts a Sequence into | |
201 // |shared_priority_queue_| and ends the Transaction. This can't happen | |
202 // if the Transaction of step 1 is still active because because there can | |
203 // only be one active Transaction per PriorityQueue at a time. | |
204 // 3. Other thread calls WakeUpOneThread(). No thread is woken up because | |
205 // |idle_worker_threads_stack_| is empty. | |
206 // 4. This thread adds itself to |idle_worker_threads_stack_| and goes to | |
207 // sleep. No thread runs the Sequence inserted in step 2. | |
208 outer_->AddToIdleWorkerThreadsStack(worker_thread); | |
209 return nullptr; | |
210 } | |
211 | |
212 transaction->Pop(); | |
213 return sequence_and_sort_key.sequence; | |
214 } | |
215 | |
216 void SchedulerThreadPool::SchedulerWorkerThreadDelegateImpl::EnqueueSequence( | |
217 scoped_refptr<Sequence> sequence) { | |
218 enqueue_sequence_callback_.Run(std::move(sequence)); | |
219 } | |
220 | |
221 SchedulerThreadPool::SchedulerThreadPool( | |
222 const EnqueueSequenceCallback& enqueue_sequence_callback, | |
223 TaskTracker* task_tracker) | |
224 : idle_worker_threads_stack_lock_(shared_priority_queue_.container_lock()), | |
225 idle_worker_threads_stack_cv_for_testing_( | |
226 idle_worker_threads_stack_lock_.CreateConditionVariable()), | |
227 join_for_testing_returned_(true, false), | |
228 worker_thread_delegate_( | |
229 new SchedulerWorkerThreadDelegateImpl(this, | |
230 enqueue_sequence_callback)), | |
231 task_tracker_(task_tracker) { | |
232 DCHECK(task_tracker_); | |
233 } | |
234 | |
235 bool SchedulerThreadPool::Initialize(ThreadPriority thread_priority, | |
236 size_t max_threads) { | |
237 AutoSchedulerLock auto_lock(idle_worker_threads_stack_lock_); | |
238 | |
239 DCHECK(worker_threads_.empty()); | |
240 | |
241 for (size_t i = 0; i < max_threads; ++i) { | |
242 std::unique_ptr<SchedulerWorkerThread> worker_thread = | |
243 SchedulerWorkerThread::CreateSchedulerWorkerThread( | |
244 thread_priority, worker_thread_delegate_.get(), task_tracker_); | |
245 if (!worker_thread) | |
246 break; | |
247 idle_worker_threads_stack_.push(worker_thread.get()); | |
248 worker_threads_.push_back(std::move(worker_thread)); | |
249 } | |
250 | |
251 return !worker_threads_.empty(); | |
252 } | |
253 | |
254 void SchedulerThreadPool::WakeUpOneThread() { | |
255 SchedulerWorkerThread* worker_thread = PopOneIdleWorkerThread(); | |
256 if (worker_thread) | |
257 worker_thread->WakeUp(); | |
258 } | |
259 | |
260 void SchedulerThreadPool::AddToIdleWorkerThreadsStack( | |
261 SchedulerWorkerThread* worker_thread) { | |
262 AutoSchedulerLock auto_lock(idle_worker_threads_stack_lock_); | |
263 idle_worker_threads_stack_.push(worker_thread); | |
264 DCHECK_LE(idle_worker_threads_stack_.size(), worker_threads_.size()); | |
265 | |
266 if (idle_worker_threads_stack_.size() == worker_threads_.size()) | |
267 idle_worker_threads_stack_cv_for_testing_->Broadcast(); | |
268 } | |
269 | |
270 SchedulerWorkerThread* SchedulerThreadPool::PopOneIdleWorkerThread() { | |
271 AutoSchedulerLock auto_lock(idle_worker_threads_stack_lock_); | |
272 | |
273 if (idle_worker_threads_stack_.empty()) | |
274 return nullptr; | |
275 | |
276 auto worker_thread = idle_worker_threads_stack_.top(); | |
277 idle_worker_threads_stack_.pop(); | |
278 return worker_thread; | |
279 } | |
280 | |
281 } // namespace internal | |
282 } // namespace base | |
OLD | NEW |