| 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 |