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. Not set for threads that | |
23 // don't belong to a SchedulerThreadPool. | |
24 LazyInstance<ThreadLocalPointer<const SchedulerThreadPool>>::Leaky | |
25 tls_current_thread_pool = LAZY_INSTANCE_INITIALIZER; | |
26 | |
27 // A task runner that runs tasks with the PARALLEL ExecutionMode. | |
28 class SchedulerParallelTaskRunner : public TaskRunner { | |
29 public: | |
30 SchedulerParallelTaskRunner(const TaskTraits& traits, | |
31 SchedulerTaskExecutor* executor, | |
32 TaskTracker* task_tracker) | |
33 : traits_(traits), executor_(executor), task_tracker_(task_tracker) {} | |
34 | |
35 // TaskRunner: | |
36 bool PostDelayedTask(const tracked_objects::Location& from_here, | |
37 const Closure& closure, | |
38 TimeDelta delay) override { | |
39 // Post the task as part of a one-off single-task Sequence. | |
40 return PostTaskHelper(from_here, closure, traits_, delay, | |
41 make_scoped_refptr(new Sequence), executor_, | |
42 task_tracker_); | |
43 } | |
44 | |
45 bool RunsTasksOnCurrentThread() const override { | |
46 return tls_current_thread_pool.Get().Get() == executor_; | |
47 } | |
48 | |
49 private: | |
50 ~SchedulerParallelTaskRunner() override = default; | |
51 | |
52 const TaskTraits traits_; | |
53 SchedulerTaskExecutor* const executor_; | |
54 TaskTracker* const task_tracker_; | |
55 | |
56 DISALLOW_COPY_AND_ASSIGN(SchedulerParallelTaskRunner); | |
57 }; | |
58 | |
59 } // namespace | |
60 | |
61 class SchedulerThreadPool::SchedulerWorkerThreadDelegateImpl | |
62 : public SchedulerWorkerThread::Delegate { | |
63 public: | |
64 SchedulerWorkerThreadDelegateImpl( | |
65 SchedulerThreadPool* outer, | |
66 const EnqueueSequenceCallback enqueue_sequence_callback); | |
67 ~SchedulerWorkerThreadDelegateImpl() override; | |
68 | |
69 // SchedulerWorkerThread::Delegate: | |
70 void OnMainEntry() override; | |
71 scoped_refptr<Sequence> GetWork( | |
72 SchedulerWorkerThread* worker_thread) override; | |
73 void EnqueueSequence(scoped_refptr<Sequence> sequence) override; | |
74 | |
75 private: | |
76 SchedulerThreadPool* outer_; | |
77 const EnqueueSequenceCallback enqueue_sequence_callback_; | |
78 | |
79 DISALLOW_COPY_AND_ASSIGN(SchedulerWorkerThreadDelegateImpl); | |
80 }; | |
81 | |
82 std::unique_ptr<SchedulerThreadPool> SchedulerThreadPool::CreateThreadPool( | |
83 ThreadPriority thread_priority, | |
84 size_t max_threads, | |
85 const EnqueueSequenceCallback& enqueue_sequence_callback, | |
86 TaskTracker* task_tracker) { | |
87 std::unique_ptr<SchedulerThreadPool> thread_pool( | |
88 new SchedulerThreadPool(enqueue_sequence_callback, task_tracker)); | |
89 if (thread_pool->Initialize(thread_priority, max_threads)) | |
90 return thread_pool; | |
91 return nullptr; | |
92 } | |
93 | |
94 SchedulerThreadPool::~SchedulerThreadPool() { | |
95 #if DCHECK_IS_ON() | |
96 // SchedulerThreadPool should never be deleted in production unless its | |
97 // initialization failed. | |
98 AutoSchedulerLock auto_lock(idle_worker_threads_stack_lock_); | |
99 DCHECK(join_for_testing_returned_.IsSignaled() || worker_threads_.empty()); | |
100 #endif // DCHECK_IS_ON() | |
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, this, task_tracker_)); | |
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 SchedulerThreadPool::SchedulerWorkerThreadDelegateImpl:: | |
156 SchedulerWorkerThreadDelegateImpl( | |
157 SchedulerThreadPool* outer, | |
158 const EnqueueSequenceCallback enqueue_sequence_callback) | |
danakj
2016/04/14 19:48:58
const&?
fdoray
2016/04/14 23:46:58
Done.
| |
159 : outer_(outer), enqueue_sequence_callback_(enqueue_sequence_callback) {} | |
160 | |
161 SchedulerThreadPool::SchedulerWorkerThreadDelegateImpl:: | |
162 ~SchedulerWorkerThreadDelegateImpl() = default; | |
163 | |
164 void SchedulerThreadPool::SchedulerWorkerThreadDelegateImpl::OnMainEntry() { | |
danakj
2016/04/14 19:48:58
Just in case, if these are in a DelegateImpl and n
fdoray
2016/04/14 23:46:58
These methods are in DelegateImpl instead of Sched
| |
165 DCHECK(!tls_current_thread_pool.Get().Get()); | |
166 tls_current_thread_pool.Get().Set(outer_); | |
167 } | |
168 | |
169 scoped_refptr<Sequence> | |
170 SchedulerThreadPool::SchedulerWorkerThreadDelegateImpl::GetWork( | |
171 SchedulerWorkerThread* worker_thread) { | |
172 std::unique_ptr<PriorityQueue::Transaction> transaction( | |
173 outer_->shared_priority_queue_.BeginTransaction()); | |
174 const auto sequence_and_sort_key = transaction->Peek(); | |
175 | |
176 if (sequence_and_sort_key.is_null()) { | |
177 // |transaction| is kept alive while |worker_thread| is added to | |
178 // |idle_worker_threads_stack_| to avoid this race: | |
179 // 1. This thread creates a Transaction, finds |shared_priority_queue_| | |
180 // empty and ends the Transaction. | |
181 // 2. Other thread creates a Transaction, inserts a Sequence into | |
182 // |shared_priority_queue_| and ends the Transaction. This can't happen | |
183 // if the Transaction of step 1 is still active because because there can | |
184 // only be one active Transaction per PriorityQueue at a time. | |
185 // 3. Other thread calls WakeUpOneThread(). No thread is woken up because | |
186 // |idle_worker_threads_stack_| is empty. | |
187 // 4. This thread adds itself to |idle_worker_threads_stack_| and goes to | |
188 // sleep. No thread runs the Sequence inserted in step 2. | |
189 outer_->AddToIdleWorkerThreadsStack(worker_thread); | |
190 return nullptr; | |
191 } | |
192 | |
193 transaction->Pop(); | |
194 return sequence_and_sort_key.sequence; | |
195 } | |
196 | |
197 void SchedulerThreadPool::SchedulerWorkerThreadDelegateImpl::EnqueueSequence( | |
198 scoped_refptr<Sequence> sequence) { | |
199 enqueue_sequence_callback_.Run(std::move(sequence)); | |
200 } | |
201 | |
202 SchedulerThreadPool::SchedulerThreadPool( | |
203 const EnqueueSequenceCallback& enqueue_sequence_callback, | |
204 TaskTracker* task_tracker) | |
205 : idle_worker_threads_stack_lock_(shared_priority_queue_.container_lock()), | |
206 idle_worker_threads_stack_cv_for_testing_( | |
207 idle_worker_threads_stack_lock_.CreateConditionVariable()), | |
208 join_for_testing_returned_(true, false), | |
209 worker_thread_delegate_( | |
210 new SchedulerWorkerThreadDelegateImpl(this, | |
211 enqueue_sequence_callback)), | |
212 task_tracker_(task_tracker) { | |
213 DCHECK(task_tracker_); | |
214 } | |
215 | |
216 bool SchedulerThreadPool::Initialize(ThreadPriority thread_priority, | |
217 size_t max_threads) { | |
218 AutoSchedulerLock auto_lock(idle_worker_threads_stack_lock_); | |
219 | |
220 DCHECK(worker_threads_.empty()); | |
221 | |
222 for (size_t i = 0; i < max_threads; ++i) { | |
223 std::unique_ptr<SchedulerWorkerThread> worker_thread = | |
224 SchedulerWorkerThread::CreateSchedulerWorkerThread( | |
225 thread_priority, worker_thread_delegate_.get(), task_tracker_); | |
226 if (!worker_thread) | |
227 break; | |
228 idle_worker_threads_stack_.push(worker_thread.get()); | |
229 worker_threads_.push_back(std::move(worker_thread)); | |
230 } | |
231 | |
232 return !worker_threads_.empty(); | |
233 } | |
234 | |
235 void SchedulerThreadPool::WakeUpOneThread() { | |
236 SchedulerWorkerThread* worker_thread; | |
237 { | |
robliao
2016/04/13 20:50:54
It might be cleaner to take the below and encapsul
fdoray
2016/04/13 23:13:12
Done.
gab
2016/04/14 16:11:34
I disagree, now this is one more hop for the reade
robliao
2016/04/14 16:43:38
The reason why I like the separation is that it ke
fdoray
2016/04/14 18:41:11
Keeping the code as-is because I plan to create a
| |
238 AutoSchedulerLock auto_lock(idle_worker_threads_stack_lock_); | |
239 | |
240 if (idle_worker_threads_stack_.empty()) | |
241 return; | |
242 | |
243 worker_thread = idle_worker_threads_stack_.top(); | |
244 idle_worker_threads_stack_.pop(); | |
245 } | |
246 worker_thread->WakeUp(); | |
247 } | |
248 | |
249 void SchedulerThreadPool::AddToIdleWorkerThreadsStack( | |
250 SchedulerWorkerThread* worker_thread) { | |
251 AutoSchedulerLock auto_lock(idle_worker_threads_stack_lock_); | |
252 idle_worker_threads_stack_.push(worker_thread); | |
253 DCHECK_LE(idle_worker_threads_stack_.size(), worker_threads_.size()); | |
254 | |
255 if (idle_worker_threads_stack_.size() == worker_threads_.size()) | |
256 idle_worker_threads_stack_cv_for_testing_->Broadcast(); | |
257 } | |
258 | |
259 void SchedulerThreadPool::PostTaskNow(std::unique_ptr<Task> task, | |
260 scoped_refptr<Sequence> sequence) { | |
261 const bool sequence_was_empty = PostTaskNowHelper( | |
262 std::move(task), std::move(sequence), &shared_priority_queue_); | |
263 | |
264 // No thread has already been woken up to run Tasks from |sequence| if it was | |
265 // empty before |task| was inserted into it. | |
266 if (sequence_was_empty) | |
267 WakeUpOneThread(); | |
268 } | |
269 | |
270 } // namespace internal | |
271 } // namespace base | |
OLD | NEW |