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

Side by Side Diff: base/task_scheduler/scheduler_thread_pool_impl.cc

Issue 1876363004: TaskScheduler [11] Support ExecutionMode::SINGLE_THREADED. (Closed) Base URL: https://chromium.googlesource.com/chromium/src.git@8_delayed
Patch Set: add #include <algorithm> Created 4 years, 7 months 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
OLDNEW
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_impl.h" 5 #include "base/task_scheduler/scheduler_thread_pool_impl.h"
6 6
7 #include <algorithm>
7 #include <utility> 8 #include <utility>
8 9
9 #include "base/bind.h" 10 #include "base/bind.h"
10 #include "base/bind_helpers.h" 11 #include "base/bind_helpers.h"
11 #include "base/lazy_instance.h" 12 #include "base/lazy_instance.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/single_thread_task_runner.h"
15 #include "base/task_scheduler/delayed_task_manager.h" 16 #include "base/task_scheduler/delayed_task_manager.h"
16 #include "base/task_scheduler/task_tracker.h" 17 #include "base/task_scheduler/task_tracker.h"
17 #include "base/threading/thread_local.h" 18 #include "base/threading/thread_local.h"
18 19
19 namespace base { 20 namespace base {
20 namespace internal { 21 namespace internal {
21 22
22 namespace { 23 namespace {
23 24
24 // SchedulerThreadPool that owns the current thread, if any. 25 // SchedulerThreadPool that owns the current thread, if any.
25 LazyInstance<ThreadLocalPointer<const SchedulerThreadPool>>::Leaky 26 LazyInstance<ThreadLocalPointer<const SchedulerThreadPool>>::Leaky
26 tls_current_thread_pool = LAZY_INSTANCE_INITIALIZER; 27 tls_current_thread_pool = LAZY_INSTANCE_INITIALIZER;
27 28
29 // SchedulerWorkerThread that owns the current thread, if any.
30 LazyInstance<ThreadLocalPointer<const SchedulerWorkerThread>>::Leaky
31 tls_current_worker_thread = LAZY_INSTANCE_INITIALIZER;
32
28 // A task runner that runs tasks with the PARALLEL ExecutionMode. 33 // A task runner that runs tasks with the PARALLEL ExecutionMode.
29 class SchedulerParallelTaskRunner : public TaskRunner { 34 class SchedulerParallelTaskRunner : public TaskRunner {
30 public: 35 public:
31 // Constructs a SchedulerParallelTaskRunner which can be used to post tasks so 36 // Constructs a SchedulerParallelTaskRunner which can be used to post tasks so
32 // long as |thread_pool| is alive. 37 // long as |thread_pool| is alive.
33 // TODO(robliao): Find a concrete way to manage |thread_pool|'s memory. 38 // TODO(robliao): Find a concrete way to manage |thread_pool|'s memory.
34 SchedulerParallelTaskRunner(const TaskTraits& traits, 39 SchedulerParallelTaskRunner(const TaskTraits& traits,
35 SchedulerThreadPool* thread_pool) 40 SchedulerThreadPool* thread_pool)
36 : traits_(traits), thread_pool_(thread_pool) {} 41 : traits_(traits), thread_pool_(thread_pool) {}
37 42
38 // TaskRunner: 43 // TaskRunner:
39 bool PostDelayedTask(const tracked_objects::Location& from_here, 44 bool PostDelayedTask(const tracked_objects::Location& from_here,
40 const Closure& closure, 45 const Closure& closure,
41 TimeDelta delay) override { 46 TimeDelta delay) override {
42 // Post the task as part of a one-off single-task Sequence. 47 // Post the task as part of a one-off single-task Sequence.
43 return thread_pool_->PostTaskWithSequence( 48 return thread_pool_->PostTaskWithSequence(
44 WrapUnique(new Task(from_here, closure, traits_, delay)), 49 WrapUnique(new Task(from_here, closure, traits_, delay)),
45 make_scoped_refptr(new Sequence)); 50 make_scoped_refptr(new Sequence), nullptr);
46 } 51 }
47 52
48 bool RunsTasksOnCurrentThread() const override { 53 bool RunsTasksOnCurrentThread() const override {
49 return tls_current_thread_pool.Get().Get() == thread_pool_; 54 return tls_current_thread_pool.Get().Get() == thread_pool_;
50 } 55 }
51 56
52 private: 57 private:
53 ~SchedulerParallelTaskRunner() override = default; 58 ~SchedulerParallelTaskRunner() override = default;
54 59
55 const TaskTraits traits_; 60 const TaskTraits traits_;
56 SchedulerThreadPool* const thread_pool_; 61 SchedulerThreadPool* const thread_pool_;
57 62
58 DISALLOW_COPY_AND_ASSIGN(SchedulerParallelTaskRunner); 63 DISALLOW_COPY_AND_ASSIGN(SchedulerParallelTaskRunner);
59 }; 64 };
60 65
61 // A task runner that runs tasks with the SEQUENCED ExecutionMode. 66 // A task runner that runs tasks with the SEQUENCED ExecutionMode.
62 class SchedulerSequencedTaskRunner : public SequencedTaskRunner { 67 class SchedulerSequencedTaskRunner : public SequencedTaskRunner {
63 public: 68 public:
64 // Constructs a SchedulerSequencedTaskRunner which can be used to post tasks 69 // Constructs a SchedulerSequencedTaskRunner which can be used to post tasks
65 // so long as |thread_pool| is alive. 70 // so long as |thread_pool| is alive.
66 // TODO(robliao): Find a concrete way to manage |thread_pool|'s memory. 71 // TODO(robliao): Find a concrete way to manage |thread_pool|'s memory.
67 SchedulerSequencedTaskRunner(const TaskTraits& traits, 72 SchedulerSequencedTaskRunner(const TaskTraits& traits,
68 SchedulerThreadPool* thread_pool) 73 SchedulerThreadPool* thread_pool)
69 : traits_(traits), thread_pool_(thread_pool) {} 74 : traits_(traits), thread_pool_(thread_pool) {}
70 75
71 // SequencedTaskRunner: 76 // SequencedTaskRunner:
72 bool PostDelayedTask(const tracked_objects::Location& from_here, 77 bool PostDelayedTask(const tracked_objects::Location& from_here,
73 const Closure& closure, 78 const Closure& closure,
74 TimeDelta delay) override { 79 TimeDelta delay) override {
75 // Post the task as part of |sequence|. 80 // Post the task as part of |sequence_|.
76 return thread_pool_->PostTaskWithSequence( 81 return thread_pool_->PostTaskWithSequence(
77 WrapUnique(new Task(from_here, closure, traits_, delay)), sequence_); 82 WrapUnique(new Task(from_here, closure, traits_, delay)), sequence_,
83 nullptr);
78 } 84 }
79 85
80 bool PostNonNestableDelayedTask(const tracked_objects::Location& from_here, 86 bool PostNonNestableDelayedTask(const tracked_objects::Location& from_here,
81 const Closure& closure, 87 const Closure& closure,
82 base::TimeDelta delay) override { 88 base::TimeDelta delay) override {
83 // Tasks are never nested within the task scheduler. 89 // Tasks are never nested within the task scheduler.
84 return PostDelayedTask(from_here, closure, delay); 90 return PostDelayedTask(from_here, closure, delay);
85 } 91 }
86 92
87 bool RunsTasksOnCurrentThread() const override { 93 bool RunsTasksOnCurrentThread() const override {
88 return tls_current_thread_pool.Get().Get() == thread_pool_; 94 return tls_current_thread_pool.Get().Get() == thread_pool_;
89 } 95 }
90 96
91 private: 97 private:
92 ~SchedulerSequencedTaskRunner() override = default; 98 ~SchedulerSequencedTaskRunner() override = default;
93 99
94 // Sequence for all Tasks posted through this TaskRunner. 100 // Sequence for all Tasks posted through this TaskRunner.
95 const scoped_refptr<Sequence> sequence_ = new Sequence; 101 const scoped_refptr<Sequence> sequence_ = new Sequence;
96 102
97 const TaskTraits traits_; 103 const TaskTraits traits_;
98 SchedulerThreadPool* const thread_pool_; 104 SchedulerThreadPool* const thread_pool_;
99 105
100 DISALLOW_COPY_AND_ASSIGN(SchedulerSequencedTaskRunner); 106 DISALLOW_COPY_AND_ASSIGN(SchedulerSequencedTaskRunner);
101 }; 107 };
102 108
109 // A task runner that runs tasks with the SINGLE_THREADED ExecutionMode.
110 class SchedulerSingleThreadTaskRunner : public SingleThreadTaskRunner {
111 public:
112 // Constructs a SchedulerSingleThreadTaskRunner which can be used to post
113 // tasks so long as |thread_pool| and |worker_thread| are alive.
114 // TODO(robliao): Find a concrete way to manage the memory of |thread_pool|
115 // and |worker_thread|.
116 SchedulerSingleThreadTaskRunner(const TaskTraits& traits,
117 SchedulerThreadPool* thread_pool,
118 SchedulerWorkerThread* worker_thread)
119 : traits_(traits),
120 thread_pool_(thread_pool),
121 worker_thread_(worker_thread) {}
122
123 // SingleThreadTaskRunner:
124 bool PostDelayedTask(const tracked_objects::Location& from_here,
125 const Closure& closure,
126 TimeDelta delay) override {
127 // Post the task to be executed by |worker_thread_| as part of |sequence_|.
128 return thread_pool_->PostTaskWithSequence(
129 WrapUnique(new Task(from_here, closure, traits_, delay)), sequence_,
130 worker_thread_);
131 }
132
133 bool PostNonNestableDelayedTask(const tracked_objects::Location& from_here,
134 const Closure& closure,
135 base::TimeDelta delay) override {
136 // Tasks are never nested within the task scheduler.
137 return PostDelayedTask(from_here, closure, delay);
138 }
139
140 bool RunsTasksOnCurrentThread() const override {
141 return tls_current_worker_thread.Get().Get() == worker_thread_;
142 }
143
144 private:
145 ~SchedulerSingleThreadTaskRunner() override = default;
146
147 // Sequence for all Tasks posted through this TaskRunner.
148 const scoped_refptr<Sequence> sequence_ = new Sequence;
149
150 const TaskTraits traits_;
151 SchedulerThreadPool* const thread_pool_;
152 SchedulerWorkerThread* const worker_thread_;
153
154 DISALLOW_COPY_AND_ASSIGN(SchedulerSingleThreadTaskRunner);
155 };
156
157 // Only used in DCHECKs.
158 bool ContainsWorkerThread(
159 const std::vector<std::unique_ptr<SchedulerWorkerThread>>& worker_threads,
160 const SchedulerWorkerThread* worker_thread) {
161 auto it = std::find_if(
162 worker_threads.begin(), worker_threads.end(),
163 [worker_thread](const std::unique_ptr<SchedulerWorkerThread>& i) {
164 return i.get() == worker_thread;
165 });
166 return it != worker_threads.end();
167 }
168
103 } // namespace 169 } // namespace
104 170
105 class SchedulerThreadPoolImpl::SchedulerWorkerThreadDelegateImpl 171 class SchedulerThreadPoolImpl::SchedulerWorkerThreadDelegateImpl
106 : public SchedulerWorkerThread::Delegate { 172 : public SchedulerWorkerThread::Delegate {
107 public: 173 public:
174 // |outer| owns the worker thread for which this delegate is constructed.
175 // |re_enqueue_sequence_callback| is invoked when ReEnqueueSequence() is
176 // called with a non-single-threaded Sequence. |shared_priority_queue| is a
177 // PriorityQueue whose transactions may overlap with the worker thread's
178 // single-threaded PriorityQueue's transactions.
108 SchedulerWorkerThreadDelegateImpl( 179 SchedulerWorkerThreadDelegateImpl(
109 SchedulerThreadPoolImpl* outer, 180 SchedulerThreadPoolImpl* outer,
110 const ReEnqueueSequenceCallback& re_enqueue_sequence_callback); 181 const ReEnqueueSequenceCallback& re_enqueue_sequence_callback,
182 const PriorityQueue* shared_priority_queue);
111 ~SchedulerWorkerThreadDelegateImpl() override; 183 ~SchedulerWorkerThreadDelegateImpl() override;
112 184
185 PriorityQueue* single_threaded_priority_queue() {
186 return &single_threaded_priority_queue_;
187 }
188
113 // SchedulerWorkerThread::Delegate: 189 // SchedulerWorkerThread::Delegate:
114 void OnMainEntry() override; 190 void OnMainEntry(SchedulerWorkerThread* worker_thread) override;
115 scoped_refptr<Sequence> GetWork( 191 scoped_refptr<Sequence> GetWork(
116 SchedulerWorkerThread* worker_thread) override; 192 SchedulerWorkerThread* worker_thread) override;
117 void ReEnqueueSequence(scoped_refptr<Sequence> sequence) override; 193 void ReEnqueueSequence(scoped_refptr<Sequence> sequence) override;
118 194
119 private: 195 private:
120 SchedulerThreadPoolImpl* outer_; 196 SchedulerThreadPoolImpl* outer_;
121 const ReEnqueueSequenceCallback re_enqueue_sequence_callback_; 197 const ReEnqueueSequenceCallback re_enqueue_sequence_callback_;
122 198
199 // Single-threaded PriorityQueue for the worker thread.
200 PriorityQueue single_threaded_priority_queue_;
201
202 // True if the last Sequence returned by GetWork() was extracted from
203 // |single_threaded_priority_queue_|.
204 bool last_sequence_is_single_threaded_ = false;
205
123 DISALLOW_COPY_AND_ASSIGN(SchedulerWorkerThreadDelegateImpl); 206 DISALLOW_COPY_AND_ASSIGN(SchedulerWorkerThreadDelegateImpl);
124 }; 207 };
125 208
126 SchedulerThreadPoolImpl::~SchedulerThreadPoolImpl() { 209 SchedulerThreadPoolImpl::~SchedulerThreadPoolImpl() {
127 // SchedulerThreadPool should never be deleted in production unless its 210 // SchedulerThreadPool should never be deleted in production unless its
128 // initialization failed. 211 // initialization failed.
129 DCHECK(join_for_testing_returned_.IsSignaled() || worker_threads_.empty()); 212 DCHECK(join_for_testing_returned_.IsSignaled() || worker_threads_.empty());
130 } 213 }
131 214
215 // static
132 std::unique_ptr<SchedulerThreadPoolImpl> SchedulerThreadPoolImpl::Create( 216 std::unique_ptr<SchedulerThreadPoolImpl> SchedulerThreadPoolImpl::Create(
133 ThreadPriority thread_priority, 217 ThreadPriority thread_priority,
134 size_t max_threads, 218 size_t max_threads,
135 const ReEnqueueSequenceCallback& re_enqueue_sequence_callback, 219 const ReEnqueueSequenceCallback& re_enqueue_sequence_callback,
136 TaskTracker* task_tracker, 220 TaskTracker* task_tracker,
137 DelayedTaskManager* delayed_task_manager) { 221 DelayedTaskManager* delayed_task_manager) {
138 std::unique_ptr<SchedulerThreadPoolImpl> thread_pool( 222 std::unique_ptr<SchedulerThreadPoolImpl> thread_pool(
139 new SchedulerThreadPoolImpl(task_tracker, delayed_task_manager)); 223 new SchedulerThreadPoolImpl(task_tracker, delayed_task_manager));
140 if (thread_pool->Initialize(thread_priority, max_threads, 224 if (thread_pool->Initialize(thread_priority, max_threads,
141 re_enqueue_sequence_callback)) { 225 re_enqueue_sequence_callback)) {
(...skipping 19 matching lines...) Expand all
161 scoped_refptr<TaskRunner> SchedulerThreadPoolImpl::CreateTaskRunnerWithTraits( 245 scoped_refptr<TaskRunner> SchedulerThreadPoolImpl::CreateTaskRunnerWithTraits(
162 const TaskTraits& traits, 246 const TaskTraits& traits,
163 ExecutionMode execution_mode) { 247 ExecutionMode execution_mode) {
164 switch (execution_mode) { 248 switch (execution_mode) {
165 case ExecutionMode::PARALLEL: 249 case ExecutionMode::PARALLEL:
166 return make_scoped_refptr(new SchedulerParallelTaskRunner(traits, this)); 250 return make_scoped_refptr(new SchedulerParallelTaskRunner(traits, this));
167 251
168 case ExecutionMode::SEQUENCED: 252 case ExecutionMode::SEQUENCED:
169 return make_scoped_refptr(new SchedulerSequencedTaskRunner(traits, this)); 253 return make_scoped_refptr(new SchedulerSequencedTaskRunner(traits, this));
170 254
171 case ExecutionMode::SINGLE_THREADED: 255 case ExecutionMode::SINGLE_THREADED: {
172 // TODO(fdoray): Support SINGLE_THREADED TaskRunners. 256 // TODO(fdoray): Find a way to take load into account when assigning a
173 NOTREACHED(); 257 // SchedulerWorkerThread to a SingleThreadTaskRunner. Also, this code
174 return nullptr; 258 // assumes that all SchedulerWorkerThreads are alive. Eventually, we might
259 // decide to tear down threads that haven't run tasks for a long time.
260 size_t worker_thread_index;
261 {
262 AutoSchedulerLock auto_lock(next_worker_thread_index_lock_);
263 worker_thread_index = next_worker_thread_index_;
264 next_worker_thread_index_ =
265 (next_worker_thread_index_ + 1) % worker_threads_.size();
266 }
267 return make_scoped_refptr(new SchedulerSingleThreadTaskRunner(
268 traits, this, worker_threads_[worker_thread_index].get()));
269 }
175 } 270 }
176 271
177 NOTREACHED(); 272 NOTREACHED();
178 return nullptr; 273 return nullptr;
179 } 274 }
180 275
181 void SchedulerThreadPoolImpl::ReEnqueueSequence( 276 void SchedulerThreadPoolImpl::ReEnqueueSequence(
182 scoped_refptr<Sequence> sequence, 277 scoped_refptr<Sequence> sequence,
183 const SequenceSortKey& sequence_sort_key) { 278 const SequenceSortKey& sequence_sort_key) {
184 shared_priority_queue_.BeginTransaction()->Push( 279 shared_priority_queue_.BeginTransaction()->Push(
185 WrapUnique(new PriorityQueue::SequenceAndSortKey(std::move(sequence), 280 WrapUnique(new PriorityQueue::SequenceAndSortKey(std::move(sequence),
186 sequence_sort_key))); 281 sequence_sort_key)));
187 282
188 // The thread calling this method just ran a Task from |sequence| and will 283 // The thread calling this method just ran a Task from |sequence| and will
189 // soon try to get another Sequence from which to run a Task. If the thread 284 // soon try to get another Sequence from which to run a Task. If the thread
190 // belongs to this pool, it will get that Sequence from 285 // belongs to this pool, it will get that Sequence from
191 // |shared_priority_queue_|. When that's the case, there is no need to wake up 286 // |shared_priority_queue_|. When that's the case, there is no need to wake up
192 // another thread after |sequence| is inserted in |shared_priority_queue_|. If 287 // another thread after |sequence| is inserted in |shared_priority_queue_|. If
193 // we did wake up another thread, we would waste resources by having more 288 // we did wake up another thread, we would waste resources by having more
194 // threads trying to get a Sequence from |shared_priority_queue_| than the 289 // threads trying to get a Sequence from |shared_priority_queue_| than the
195 // number of Sequences in it. 290 // number of Sequences in it.
196 if (tls_current_thread_pool.Get().Get() != this) 291 if (tls_current_thread_pool.Get().Get() != this)
197 WakeUpOneThread(); 292 WakeUpOneThread();
198 } 293 }
199 294
200 bool SchedulerThreadPoolImpl::PostTaskWithSequence( 295 bool SchedulerThreadPoolImpl::PostTaskWithSequence(
201 std::unique_ptr<Task> task, 296 std::unique_ptr<Task> task,
202 scoped_refptr<Sequence> sequence) { 297 scoped_refptr<Sequence> sequence,
298 SchedulerWorkerThread* worker_thread) {
203 DCHECK(task); 299 DCHECK(task);
204 DCHECK(sequence); 300 DCHECK(sequence);
301 DCHECK(!worker_thread ||
302 ContainsWorkerThread(worker_threads_, worker_thread));
205 303
206 if (!task_tracker_->WillPostTask(task.get())) 304 if (!task_tracker_->WillPostTask(task.get()))
207 return false; 305 return false;
208 306
209 if (task->delayed_run_time.is_null()) { 307 if (task->delayed_run_time.is_null()) {
210 PostTaskWithSequenceNow(std::move(task), std::move(sequence)); 308 PostTaskWithSequenceNow(std::move(task), std::move(sequence),
309 worker_thread);
211 } else { 310 } else {
212 delayed_task_manager_->AddDelayedTask(std::move(task), std::move(sequence), 311 delayed_task_manager_->AddDelayedTask(std::move(task), std::move(sequence),
213 this); 312 worker_thread, this);
214 } 313 }
215 314
216 return true; 315 return true;
217 } 316 }
218 317
219 void SchedulerThreadPoolImpl::PostTaskWithSequenceNow( 318 void SchedulerThreadPoolImpl::PostTaskWithSequenceNow(
220 std::unique_ptr<Task> task, 319 std::unique_ptr<Task> task,
221 scoped_refptr<Sequence> sequence) { 320 scoped_refptr<Sequence> sequence,
321 SchedulerWorkerThread* worker_thread) {
222 DCHECK(task); 322 DCHECK(task);
223 DCHECK(sequence); 323 DCHECK(sequence);
324 DCHECK(!worker_thread ||
325 ContainsWorkerThread(worker_threads_, worker_thread));
224 326
225 // Confirm that |task| is ready to run (its delayed run time is either null or 327 // Confirm that |task| is ready to run (its delayed run time is either null or
226 // in the past). 328 // in the past).
227 DCHECK_LE(task->delayed_run_time, delayed_task_manager_->Now()); 329 DCHECK_LE(task->delayed_run_time, delayed_task_manager_->Now());
228 330
331 // Because |worker_thread| belongs to this thread pool, we know that the type
332 // of its delegate is SchedulerWorkerThreadDelegateImpl.
333 PriorityQueue* const priority_queue =
334 worker_thread
335 ? static_cast<SchedulerWorkerThreadDelegateImpl*>(
336 worker_thread->delegate())
337 ->single_threaded_priority_queue()
338 : &shared_priority_queue_;
339 DCHECK(priority_queue);
340
229 const bool sequence_was_empty = sequence->PushTask(std::move(task)); 341 const bool sequence_was_empty = sequence->PushTask(std::move(task));
230 if (sequence_was_empty) { 342 if (sequence_was_empty) {
231 // Insert |sequence| in |shared_priority_queue_| if it was empty before 343 // Insert |sequence| in |priority_queue| if it was empty before |task| was
232 // |task| was inserted into it. Otherwise, one of these must be true: 344 // inserted into it. Otherwise, one of these must be true:
233 // - |sequence| is already in a PriorityQueue (not necessarily 345 // - |sequence| is already in a PriorityQueue (not necessarily
234 // |shared_priority_queue_|), or, 346 // |shared_priority_queue_|), or,
235 // - A worker thread is running a Task from |sequence|. It will insert 347 // - A worker thread is running a Task from |sequence|. It will insert
236 // |sequence| in a PriorityQueue once it's done running the Task. 348 // |sequence| in a PriorityQueue once it's done running the Task.
237 const auto sequence_sort_key = sequence->GetSortKey(); 349 const auto sequence_sort_key = sequence->GetSortKey();
238 shared_priority_queue_.BeginTransaction()->Push( 350 priority_queue->BeginTransaction()->Push(
239 WrapUnique(new PriorityQueue::SequenceAndSortKey(std::move(sequence), 351 WrapUnique(new PriorityQueue::SequenceAndSortKey(std::move(sequence),
240 sequence_sort_key))); 352 sequence_sort_key)));
241 353
242 // Wake up a worker thread to process |sequence|. 354 // Wake up a worker thread to process |sequence|.
243 WakeUpOneThread(); 355 if (worker_thread)
356 worker_thread->WakeUp();
357 else
358 WakeUpOneThread();
244 } 359 }
245 } 360 }
246 361
247 SchedulerThreadPoolImpl::SchedulerWorkerThreadDelegateImpl:: 362 SchedulerThreadPoolImpl::SchedulerWorkerThreadDelegateImpl::
248 SchedulerWorkerThreadDelegateImpl( 363 SchedulerWorkerThreadDelegateImpl(
249 SchedulerThreadPoolImpl* outer, 364 SchedulerThreadPoolImpl* outer,
250 const ReEnqueueSequenceCallback& re_enqueue_sequence_callback) 365 const ReEnqueueSequenceCallback& re_enqueue_sequence_callback,
366 const PriorityQueue* shared_priority_queue)
251 : outer_(outer), 367 : outer_(outer),
252 re_enqueue_sequence_callback_(re_enqueue_sequence_callback) {} 368 re_enqueue_sequence_callback_(re_enqueue_sequence_callback),
369 single_threaded_priority_queue_(shared_priority_queue) {}
253 370
254 SchedulerThreadPoolImpl::SchedulerWorkerThreadDelegateImpl:: 371 SchedulerThreadPoolImpl::SchedulerWorkerThreadDelegateImpl::
255 ~SchedulerWorkerThreadDelegateImpl() = default; 372 ~SchedulerWorkerThreadDelegateImpl() = default;
256 373
257 void SchedulerThreadPoolImpl::SchedulerWorkerThreadDelegateImpl::OnMainEntry() { 374 void SchedulerThreadPoolImpl::SchedulerWorkerThreadDelegateImpl::OnMainEntry(
375 SchedulerWorkerThread* worker_thread) {
376 #if DCHECK_IS_ON()
377 // Wait for |outer_->threads_created_| to avoid traversing
378 // |outer_->worker_threads_| while it is being filled by Initialize().
379 outer_->threads_created_.Wait();
380 DCHECK(ContainsWorkerThread(outer_->worker_threads_, worker_thread));
381 #endif
382
383 DCHECK(!tls_current_worker_thread.Get().Get());
258 DCHECK(!tls_current_thread_pool.Get().Get()); 384 DCHECK(!tls_current_thread_pool.Get().Get());
385 tls_current_worker_thread.Get().Set(worker_thread);
259 tls_current_thread_pool.Get().Set(outer_); 386 tls_current_thread_pool.Get().Set(outer_);
260 } 387 }
261 388
262 scoped_refptr<Sequence> 389 scoped_refptr<Sequence>
263 SchedulerThreadPoolImpl::SchedulerWorkerThreadDelegateImpl::GetWork( 390 SchedulerThreadPoolImpl::SchedulerWorkerThreadDelegateImpl::GetWork(
264 SchedulerWorkerThread* worker_thread) { 391 SchedulerWorkerThread* worker_thread) {
265 std::unique_ptr<PriorityQueue::Transaction> transaction( 392 DCHECK(ContainsWorkerThread(outer_->worker_threads_, worker_thread));
266 outer_->shared_priority_queue_.BeginTransaction());
267 const auto& sequence_and_sort_key = transaction->Peek();
268 393
269 if (sequence_and_sort_key.is_null()) { 394 scoped_refptr<Sequence> sequence;
270 // |transaction| is kept alive while |worker_thread| is added to 395 {
271 // |idle_worker_threads_stack_| to avoid this race: 396 std::unique_ptr<PriorityQueue::Transaction> shared_transaction(
272 // 1. This thread creates a Transaction, finds |shared_priority_queue_| 397 outer_->shared_priority_queue_.BeginTransaction());
273 // empty and ends the Transaction. 398 const auto& shared_sequence_and_sort_key = shared_transaction->Peek();
274 // 2. Other thread creates a Transaction, inserts a Sequence into 399
275 // |shared_priority_queue_| and ends the Transaction. This can't happen 400 std::unique_ptr<PriorityQueue::Transaction> single_threaded_transaction(
276 // if the Transaction of step 1 is still active because because there can 401 single_threaded_priority_queue_.BeginTransaction());
277 // only be one active Transaction per PriorityQueue at a time. 402 const auto& single_threaded_sequence_and_sort_key =
278 // 3. Other thread calls WakeUpOneThread(). No thread is woken up because 403 single_threaded_transaction->Peek();
279 // |idle_worker_threads_stack_| is empty. 404
280 // 4. This thread adds itself to |idle_worker_threads_stack_| and goes to 405 if (shared_sequence_and_sort_key.is_null() &&
281 // sleep. No thread runs the Sequence inserted in step 2. 406 single_threaded_sequence_and_sort_key.is_null()) {
282 outer_->AddToIdleWorkerThreadsStack(worker_thread); 407 single_threaded_transaction.reset();
283 return nullptr; 408
409 // |shared_transaction| is kept alive while |worker_thread| is added to
410 // |idle_worker_threads_stack_| to avoid this race:
411 // 1. This thread creates a Transaction, finds |shared_priority_queue_|
412 // empty and ends the Transaction.
413 // 2. Other thread creates a Transaction, inserts a Sequence into
414 // |shared_priority_queue_| and ends the Transaction. This can't happen
415 // if the Transaction of step 1 is still active because because there
416 // can only be one active Transaction per PriorityQueue at a time.
417 // 3. Other thread calls WakeUpOneThread(). No thread is woken up because
418 // |idle_worker_threads_stack_| is empty.
419 // 4. This thread adds itself to |idle_worker_threads_stack_| and goes to
420 // sleep. No thread runs the Sequence inserted in step 2.
421 outer_->AddToIdleWorkerThreadsStack(worker_thread);
422 return nullptr;
423 }
424
425 // True if both PriorityQueues have Sequences and the Sequence at the top of
426 // the shared PriorityQueue is more important.
427 const bool shared_sequence_is_more_important =
428 !shared_sequence_and_sort_key.is_null() &&
429 !single_threaded_sequence_and_sort_key.is_null() &&
430 shared_sequence_and_sort_key.sort_key >
431 single_threaded_sequence_and_sort_key.sort_key;
432
433 if (single_threaded_sequence_and_sort_key.is_null() ||
434 shared_sequence_is_more_important) {
435 sequence = shared_sequence_and_sort_key.sequence;
436 shared_transaction->Pop();
437 last_sequence_is_single_threaded_ = false;
438 } else {
439 DCHECK(!single_threaded_sequence_and_sort_key.is_null());
440 sequence = single_threaded_sequence_and_sort_key.sequence;
441 single_threaded_transaction->Pop();
442 last_sequence_is_single_threaded_ = true;
443 }
284 } 444 }
445 DCHECK(sequence);
285 446
286 scoped_refptr<Sequence> sequence = sequence_and_sort_key.sequence; 447 outer_->RemoveFromIdleWorkerThreadsStack(worker_thread);
287 transaction->Pop();
288 return sequence; 448 return sequence;
289 } 449 }
290 450
291 void SchedulerThreadPoolImpl::SchedulerWorkerThreadDelegateImpl:: 451 void SchedulerThreadPoolImpl::SchedulerWorkerThreadDelegateImpl::
292 ReEnqueueSequence(scoped_refptr<Sequence> sequence) { 452 ReEnqueueSequence(scoped_refptr<Sequence> sequence) {
293 re_enqueue_sequence_callback_.Run(std::move(sequence)); 453 if (last_sequence_is_single_threaded_) {
454 // A single-threaded Sequence is always re-enqueued in the single-threaded
455 // PriorityQueue from which it was extracted.
456 const SequenceSortKey sequence_sort_key = sequence->GetSortKey();
457 single_threaded_priority_queue_.BeginTransaction()->Push(
458 WrapUnique(new PriorityQueue::SequenceAndSortKey(std::move(sequence),
459 sequence_sort_key)));
460 } else {
461 // |re_enqueue_sequence_callback_| will determine in which PriorityQueue
462 // |sequence| must be enqueued.
463 re_enqueue_sequence_callback_.Run(std::move(sequence));
464 }
294 } 465 }
295 466
296 SchedulerThreadPoolImpl::SchedulerThreadPoolImpl( 467 SchedulerThreadPoolImpl::SchedulerThreadPoolImpl(
297 TaskTracker* task_tracker, 468 TaskTracker* task_tracker,
298 DelayedTaskManager* delayed_task_manager) 469 DelayedTaskManager* delayed_task_manager)
299 : idle_worker_threads_stack_lock_(shared_priority_queue_.container_lock()), 470 : idle_worker_threads_stack_lock_(shared_priority_queue_.container_lock()),
300 idle_worker_threads_stack_cv_for_testing_( 471 idle_worker_threads_stack_cv_for_testing_(
301 idle_worker_threads_stack_lock_.CreateConditionVariable()), 472 idle_worker_threads_stack_lock_.CreateConditionVariable()),
302 join_for_testing_returned_(true, false), 473 join_for_testing_returned_(true, false),
474 #if DCHECK_IS_ON()
475 threads_created_(true, false),
476 #endif
303 task_tracker_(task_tracker), 477 task_tracker_(task_tracker),
304 delayed_task_manager_(delayed_task_manager) { 478 delayed_task_manager_(delayed_task_manager) {
305 DCHECK(task_tracker_); 479 DCHECK(task_tracker_);
306 DCHECK(delayed_task_manager_); 480 DCHECK(delayed_task_manager_);
307 } 481 }
308 482
309 bool SchedulerThreadPoolImpl::Initialize( 483 bool SchedulerThreadPoolImpl::Initialize(
310 ThreadPriority thread_priority, 484 ThreadPriority thread_priority,
311 size_t max_threads, 485 size_t max_threads,
312 const ReEnqueueSequenceCallback& re_enqueue_sequence_callback) { 486 const ReEnqueueSequenceCallback& re_enqueue_sequence_callback) {
313 AutoSchedulerLock auto_lock(idle_worker_threads_stack_lock_); 487 AutoSchedulerLock auto_lock(idle_worker_threads_stack_lock_);
314 488
315 DCHECK(worker_threads_.empty()); 489 DCHECK(worker_threads_.empty());
316 490
317 for (size_t i = 0; i < max_threads; ++i) { 491 for (size_t i = 0; i < max_threads; ++i) {
318 std::unique_ptr<SchedulerWorkerThread> worker_thread = 492 std::unique_ptr<SchedulerWorkerThread> worker_thread =
319 SchedulerWorkerThread::Create( 493 SchedulerWorkerThread::Create(
320 thread_priority, WrapUnique(new SchedulerWorkerThreadDelegateImpl( 494 thread_priority,
321 this, re_enqueue_sequence_callback)), 495 WrapUnique(new SchedulerWorkerThreadDelegateImpl(
496 this, re_enqueue_sequence_callback, &shared_priority_queue_)),
322 task_tracker_); 497 task_tracker_);
323 if (!worker_thread) 498 if (!worker_thread)
324 break; 499 break;
325 idle_worker_threads_stack_.Push(worker_thread.get()); 500 idle_worker_threads_stack_.Push(worker_thread.get());
326 worker_threads_.push_back(std::move(worker_thread)); 501 worker_threads_.push_back(std::move(worker_thread));
327 } 502 }
328 503
504 #if DCHECK_IS_ON()
505 threads_created_.Signal();
506 #endif
507
329 return !worker_threads_.empty(); 508 return !worker_threads_.empty();
330 } 509 }
331 510
332 void SchedulerThreadPoolImpl::WakeUpOneThread() { 511 void SchedulerThreadPoolImpl::WakeUpOneThread() {
333 SchedulerWorkerThread* worker_thread; 512 SchedulerWorkerThread* worker_thread;
334 { 513 {
335 AutoSchedulerLock auto_lock(idle_worker_threads_stack_lock_); 514 AutoSchedulerLock auto_lock(idle_worker_threads_stack_lock_);
336 worker_thread = idle_worker_threads_stack_.Pop(); 515 worker_thread = idle_worker_threads_stack_.Pop();
337 } 516 }
338 if (worker_thread) 517 if (worker_thread)
339 worker_thread->WakeUp(); 518 worker_thread->WakeUp();
340 } 519 }
341 520
342 void SchedulerThreadPoolImpl::AddToIdleWorkerThreadsStack( 521 void SchedulerThreadPoolImpl::AddToIdleWorkerThreadsStack(
343 SchedulerWorkerThread* worker_thread) { 522 SchedulerWorkerThread* worker_thread) {
344 AutoSchedulerLock auto_lock(idle_worker_threads_stack_lock_); 523 AutoSchedulerLock auto_lock(idle_worker_threads_stack_lock_);
345 idle_worker_threads_stack_.Push(worker_thread); 524 idle_worker_threads_stack_.Push(worker_thread);
346 DCHECK_LE(idle_worker_threads_stack_.Size(), worker_threads_.size()); 525 DCHECK_LE(idle_worker_threads_stack_.Size(), worker_threads_.size());
347 526
348 if (idle_worker_threads_stack_.Size() == worker_threads_.size()) 527 if (idle_worker_threads_stack_.Size() == worker_threads_.size())
349 idle_worker_threads_stack_cv_for_testing_->Broadcast(); 528 idle_worker_threads_stack_cv_for_testing_->Broadcast();
350 } 529 }
351 530
531 void SchedulerThreadPoolImpl::RemoveFromIdleWorkerThreadsStack(
532 SchedulerWorkerThread* worker_thread) {
533 AutoSchedulerLock auto_lock(idle_worker_threads_stack_lock_);
534 idle_worker_threads_stack_.Remove(worker_thread);
535 }
536
352 } // namespace internal 537 } // namespace internal
353 } // namespace base 538 } // namespace base
OLDNEW
« no previous file with comments | « base/task_scheduler/scheduler_thread_pool_impl.h ('k') | base/task_scheduler/scheduler_thread_pool_impl_unittest.cc » ('j') | no next file with comments »

Powered by Google App Engine
This is Rietveld 408576698