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

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

Issue 1906083002: TaskScheduler: Remove base/task_scheduler/utils.h/.cc (Closed) Base URL: https://chromium.googlesource.com/chromium/src.git@sched_2_stack
Patch Set: Created 4 years, 8 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.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
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
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
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
OLDNEW

Powered by Google App Engine
This is Rietveld 408576698