| 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/sequenced_task_runner.h" | |
| 15 #include "base/task_scheduler/utils.h" | |
| 16 #include "base/threading/thread_local.h" | |
| 17 | |
| 18 namespace base { | |
| 19 namespace internal { | |
| 20 | |
| 21 namespace { | |
| 22 | |
| 23 // SchedulerThreadPool that owns the current thread, if any. | |
| 24 LazyInstance<ThreadLocalPointer<const SchedulerTaskExecutor>>::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 // Constructs a SchedulerParallelTaskRunner which can be used to post tasks so | |
| 31 // long as |executor| is alive. | |
| 32 // TODO(robliao): Find a concrete way to manage |executor|'s memory. | |
| 33 SchedulerParallelTaskRunner(const TaskTraits& traits, | |
| 34 SchedulerTaskExecutor* executor, | |
| 35 TaskTracker* task_tracker, | |
| 36 DelayedTaskManager* delayed_task_manager) | |
| 37 : traits_(traits), | |
| 38 executor_(executor), | |
| 39 task_tracker_(task_tracker), | |
| 40 delayed_task_manager_(delayed_task_manager) {} | |
| 41 | |
| 42 // TaskRunner: | |
| 43 bool PostDelayedTask(const tracked_objects::Location& from_here, | |
| 44 const Closure& closure, | |
| 45 TimeDelta delay) override { | |
| 46 // Post the task as part of a one-off single-task Sequence. | |
| 47 return PostTaskToExecutor( | |
| 48 WrapUnique( | |
| 49 new Task(from_here, closure, traits_, | |
| 50 delay.is_zero() ? TimeTicks() : TimeTicks::Now() + delay)), | |
| 51 make_scoped_refptr(new Sequence), executor_, task_tracker_, | |
| 52 delayed_task_manager_); | |
| 53 } | |
| 54 | |
| 55 bool RunsTasksOnCurrentThread() const override { | |
| 56 return tls_current_thread_pool.Get().Get() == executor_; | |
| 57 } | |
| 58 | |
| 59 private: | |
| 60 ~SchedulerParallelTaskRunner() override = default; | |
| 61 | |
| 62 const TaskTraits traits_; | |
| 63 SchedulerTaskExecutor* const executor_; | |
| 64 TaskTracker* const task_tracker_; | |
| 65 DelayedTaskManager* const delayed_task_manager_; | |
| 66 | |
| 67 DISALLOW_COPY_AND_ASSIGN(SchedulerParallelTaskRunner); | |
| 68 }; | |
| 69 | |
| 70 // A task runner that runs tasks with the SEQUENCED ExecutionMode. | |
| 71 class SchedulerSequencedTaskRunner : public SequencedTaskRunner { | |
| 72 public: | |
| 73 // Constructs a SchedulerParallelTaskRunner which can be used to post tasks so | |
| 74 // long as |executor| is alive. | |
| 75 // TODO(robliao): Find a concrete way to manage |executor|'s memory. | |
| 76 SchedulerSequencedTaskRunner(const TaskTraits& traits, | |
| 77 SchedulerTaskExecutor* executor, | |
| 78 TaskTracker* task_tracker, | |
| 79 DelayedTaskManager* delayed_task_manager) | |
| 80 : traits_(traits), | |
| 81 executor_(executor), | |
| 82 task_tracker_(task_tracker), | |
| 83 delayed_task_manager_(delayed_task_manager) {} | |
| 84 | |
| 85 // SequencedTaskRunner: | |
| 86 bool PostDelayedTask(const tracked_objects::Location& from_here, | |
| 87 const Closure& closure, | |
| 88 TimeDelta delay) override { | |
| 89 // Post the task as part of |sequence|. | |
| 90 return PostTaskToExecutor( | |
| 91 WrapUnique( | |
| 92 new Task(from_here, closure, traits_, | |
| 93 delay.is_zero() ? TimeTicks() : TimeTicks::Now() + delay)), | |
| 94 sequence_, executor_, task_tracker_, delayed_task_manager_); | |
| 95 } | |
| 96 | |
| 97 bool PostNonNestableDelayedTask(const tracked_objects::Location& from_here, | |
| 98 const Closure& closure, | |
| 99 base::TimeDelta delay) override { | |
| 100 // Tasks are never nested within the task scheduler. | |
| 101 return PostDelayedTask(from_here, closure, delay); | |
| 102 } | |
| 103 | |
| 104 bool RunsTasksOnCurrentThread() const override { | |
| 105 return tls_current_thread_pool.Get().Get() == executor_; | |
| 106 } | |
| 107 | |
| 108 private: | |
| 109 ~SchedulerSequencedTaskRunner() override = default; | |
| 110 | |
| 111 // Sequence for all Tasks posted through this TaskRunner. | |
| 112 const scoped_refptr<Sequence> sequence_ = new Sequence; | |
| 113 | |
| 114 const TaskTraits traits_; | |
| 115 SchedulerTaskExecutor* const executor_; | |
| 116 TaskTracker* const task_tracker_; | |
| 117 DelayedTaskManager* const delayed_task_manager_; | |
| 118 | |
| 119 DISALLOW_COPY_AND_ASSIGN(SchedulerSequencedTaskRunner); | |
| 120 }; | |
| 121 | |
| 122 } // namespace | |
| 123 | |
| 124 class SchedulerThreadPool::SchedulerWorkerThreadDelegateImpl | |
| 125 : public SchedulerWorkerThread::Delegate { | |
| 126 public: | |
| 127 SchedulerWorkerThreadDelegateImpl( | |
| 128 SchedulerThreadPool* outer, | |
| 129 const EnqueueSequenceCallback& enqueue_sequence_callback); | |
| 130 ~SchedulerWorkerThreadDelegateImpl() override; | |
| 131 | |
| 132 // SchedulerWorkerThread::Delegate: | |
| 133 void OnMainEntry() override; | |
| 134 scoped_refptr<Sequence> GetWork( | |
| 135 SchedulerWorkerThread* worker_thread) override; | |
| 136 void EnqueueSequence(scoped_refptr<Sequence> sequence) override; | |
| 137 | |
| 138 private: | |
| 139 SchedulerThreadPool* outer_; | |
| 140 const EnqueueSequenceCallback enqueue_sequence_callback_; | |
| 141 | |
| 142 DISALLOW_COPY_AND_ASSIGN(SchedulerWorkerThreadDelegateImpl); | |
| 143 }; | |
| 144 | |
| 145 SchedulerThreadPool::~SchedulerThreadPool() { | |
| 146 // SchedulerThreadPool should never be deleted in production unless its | |
| 147 // initialization failed. | |
| 148 DCHECK(join_for_testing_returned_.IsSignaled() || worker_threads_.empty()); | |
| 149 } | |
| 150 | |
| 151 std::unique_ptr<SchedulerThreadPool> SchedulerThreadPool::CreateThreadPool( | |
| 152 ThreadPriority thread_priority, | |
| 153 size_t max_threads, | |
| 154 const EnqueueSequenceCallback& enqueue_sequence_callback, | |
| 155 TaskTracker* task_tracker, | |
| 156 DelayedTaskManager* delayed_task_manager) { | |
| 157 std::unique_ptr<SchedulerThreadPool> thread_pool(new SchedulerThreadPool( | |
| 158 enqueue_sequence_callback, task_tracker, delayed_task_manager)); | |
| 159 if (thread_pool->Initialize(thread_priority, max_threads)) | |
| 160 return thread_pool; | |
| 161 return nullptr; | |
| 162 } | |
| 163 | |
| 164 scoped_refptr<TaskRunner> SchedulerThreadPool::CreateTaskRunnerWithTraits( | |
| 165 const TaskTraits& traits, | |
| 166 ExecutionMode execution_mode) { | |
| 167 switch (execution_mode) { | |
| 168 case ExecutionMode::PARALLEL: | |
| 169 return make_scoped_refptr(new SchedulerParallelTaskRunner( | |
| 170 traits, this, task_tracker_, delayed_task_manager_)); | |
| 171 | |
| 172 case ExecutionMode::SEQUENCED: | |
| 173 return make_scoped_refptr(new SchedulerSequencedTaskRunner( | |
| 174 traits, this, task_tracker_, delayed_task_manager_)); | |
| 175 | |
| 176 case ExecutionMode::SINGLE_THREADED: | |
| 177 // TODO(fdoray): Support SINGLE_THREADED TaskRunners. | |
| 178 NOTREACHED(); | |
| 179 return nullptr; | |
| 180 } | |
| 181 | |
| 182 NOTREACHED(); | |
| 183 return nullptr; | |
| 184 } | |
| 185 | |
| 186 void SchedulerThreadPool::EnqueueSequence( | |
| 187 scoped_refptr<Sequence> sequence, | |
| 188 const SequenceSortKey& sequence_sort_key) { | |
| 189 shared_priority_queue_.BeginTransaction()->Push( | |
| 190 WrapUnique(new PriorityQueue::SequenceAndSortKey(std::move(sequence), | |
| 191 sequence_sort_key))); | |
| 192 | |
| 193 // The thread calling this method just ran a Task from |sequence| and will | |
| 194 // soon try to get another Sequence from which to run a Task. If the thread | |
| 195 // belongs to this pool, it will get that Sequence from | |
| 196 // |shared_priority_queue_|. When that's the case, there is no need to wake up | |
| 197 // another thread after |sequence| is inserted in |shared_priority_queue_|. If | |
| 198 // we did wake up another thread, we would waste resources by having more | |
| 199 // threads trying to get a Sequence from |shared_priority_queue_| than the | |
| 200 // number of Sequences in it. | |
| 201 if (tls_current_thread_pool.Get().Get() != this) | |
| 202 WakeUpOneThread(); | |
| 203 } | |
| 204 | |
| 205 void SchedulerThreadPool::WaitForAllWorkerThreadsIdleForTesting() { | |
| 206 AutoSchedulerLock auto_lock(idle_worker_threads_stack_lock_); | |
| 207 while (idle_worker_threads_stack_.Size() < worker_threads_.size()) | |
| 208 idle_worker_threads_stack_cv_for_testing_->Wait(); | |
| 209 } | |
| 210 | |
| 211 void SchedulerThreadPool::JoinForTesting() { | |
| 212 for (const auto& worker_thread : worker_threads_) | |
| 213 worker_thread->JoinForTesting(); | |
| 214 | |
| 215 DCHECK(!join_for_testing_returned_.IsSignaled()); | |
| 216 join_for_testing_returned_.Signal(); | |
| 217 } | |
| 218 | |
| 219 void SchedulerThreadPool::PostTaskWithSequence( | |
| 220 std::unique_ptr<Task> task, | |
| 221 scoped_refptr<Sequence> sequence) { | |
| 222 DCHECK(task); | |
| 223 DCHECK(sequence); | |
| 224 | |
| 225 const bool sequence_was_empty = AddTaskToSequenceAndPriorityQueue( | |
| 226 std::move(task), std::move(sequence), &shared_priority_queue_); | |
| 227 | |
| 228 // No thread has already been woken up to run Tasks from |sequence| if it was | |
| 229 // empty before |task| was inserted into it. | |
| 230 if (sequence_was_empty) | |
| 231 WakeUpOneThread(); | |
| 232 } | |
| 233 | |
| 234 SchedulerThreadPool::SchedulerWorkerThreadDelegateImpl:: | |
| 235 SchedulerWorkerThreadDelegateImpl( | |
| 236 SchedulerThreadPool* outer, | |
| 237 const EnqueueSequenceCallback& enqueue_sequence_callback) | |
| 238 : outer_(outer), enqueue_sequence_callback_(enqueue_sequence_callback) {} | |
| 239 | |
| 240 SchedulerThreadPool::SchedulerWorkerThreadDelegateImpl:: | |
| 241 ~SchedulerWorkerThreadDelegateImpl() = default; | |
| 242 | |
| 243 void SchedulerThreadPool::SchedulerWorkerThreadDelegateImpl::OnMainEntry() { | |
| 244 DCHECK(!tls_current_thread_pool.Get().Get()); | |
| 245 tls_current_thread_pool.Get().Set(outer_); | |
| 246 } | |
| 247 | |
| 248 scoped_refptr<Sequence> | |
| 249 SchedulerThreadPool::SchedulerWorkerThreadDelegateImpl::GetWork( | |
| 250 SchedulerWorkerThread* worker_thread) { | |
| 251 std::unique_ptr<PriorityQueue::Transaction> transaction( | |
| 252 outer_->shared_priority_queue_.BeginTransaction()); | |
| 253 const auto& sequence_and_sort_key = transaction->Peek(); | |
| 254 | |
| 255 if (sequence_and_sort_key.is_null()) { | |
| 256 // |transaction| is kept alive while |worker_thread| is added to | |
| 257 // |idle_worker_threads_stack_| to avoid this race: | |
| 258 // 1. This thread creates a Transaction, finds |shared_priority_queue_| | |
| 259 // empty and ends the Transaction. | |
| 260 // 2. Other thread creates a Transaction, inserts a Sequence into | |
| 261 // |shared_priority_queue_| and ends the Transaction. This can't happen | |
| 262 // if the Transaction of step 1 is still active because because there can | |
| 263 // only be one active Transaction per PriorityQueue at a time. | |
| 264 // 3. Other thread calls WakeUpOneThread(). No thread is woken up because | |
| 265 // |idle_worker_threads_stack_| is empty. | |
| 266 // 4. This thread adds itself to |idle_worker_threads_stack_| and goes to | |
| 267 // sleep. No thread runs the Sequence inserted in step 2. | |
| 268 outer_->AddToIdleWorkerThreadsStack(worker_thread); | |
| 269 return nullptr; | |
| 270 } | |
| 271 | |
| 272 scoped_refptr<Sequence> sequence = sequence_and_sort_key.sequence; | |
| 273 transaction->Pop(); | |
| 274 return sequence; | |
| 275 } | |
| 276 | |
| 277 void SchedulerThreadPool::SchedulerWorkerThreadDelegateImpl::EnqueueSequence( | |
| 278 scoped_refptr<Sequence> sequence) { | |
| 279 enqueue_sequence_callback_.Run(std::move(sequence)); | |
| 280 } | |
| 281 | |
| 282 SchedulerThreadPool::SchedulerThreadPool( | |
| 283 const EnqueueSequenceCallback& enqueue_sequence_callback, | |
| 284 TaskTracker* task_tracker, | |
| 285 DelayedTaskManager* delayed_task_manager) | |
| 286 : idle_worker_threads_stack_lock_(shared_priority_queue_.container_lock()), | |
| 287 idle_worker_threads_stack_cv_for_testing_( | |
| 288 idle_worker_threads_stack_lock_.CreateConditionVariable()), | |
| 289 join_for_testing_returned_(true, false), | |
| 290 worker_thread_delegate_( | |
| 291 new SchedulerWorkerThreadDelegateImpl(this, | |
| 292 enqueue_sequence_callback)), | |
| 293 task_tracker_(task_tracker), | |
| 294 delayed_task_manager_(delayed_task_manager) { | |
| 295 DCHECK(task_tracker_); | |
| 296 DCHECK(delayed_task_manager_); | |
| 297 } | |
| 298 | |
| 299 bool SchedulerThreadPool::Initialize(ThreadPriority thread_priority, | |
| 300 size_t max_threads) { | |
| 301 AutoSchedulerLock auto_lock(idle_worker_threads_stack_lock_); | |
| 302 | |
| 303 DCHECK(worker_threads_.empty()); | |
| 304 | |
| 305 for (size_t i = 0; i < max_threads; ++i) { | |
| 306 std::unique_ptr<SchedulerWorkerThread> worker_thread = | |
| 307 SchedulerWorkerThread::CreateSchedulerWorkerThread( | |
| 308 thread_priority, worker_thread_delegate_.get(), task_tracker_); | |
| 309 if (!worker_thread) | |
| 310 break; | |
| 311 idle_worker_threads_stack_.Push(worker_thread.get()); | |
| 312 worker_threads_.push_back(std::move(worker_thread)); | |
| 313 } | |
| 314 | |
| 315 return !worker_threads_.empty(); | |
| 316 } | |
| 317 | |
| 318 void SchedulerThreadPool::WakeUpOneThread() { | |
| 319 SchedulerWorkerThread* worker_thread; | |
| 320 { | |
| 321 AutoSchedulerLock auto_lock(idle_worker_threads_stack_lock_); | |
| 322 worker_thread = idle_worker_threads_stack_.Pop(); | |
| 323 } | |
| 324 if (worker_thread) | |
| 325 worker_thread->WakeUp(); | |
| 326 } | |
| 327 | |
| 328 void SchedulerThreadPool::AddToIdleWorkerThreadsStack( | |
| 329 SchedulerWorkerThread* worker_thread) { | |
| 330 AutoSchedulerLock auto_lock(idle_worker_threads_stack_lock_); | |
| 331 idle_worker_threads_stack_.Push(worker_thread); | |
| 332 DCHECK_LE(idle_worker_threads_stack_.Size(), worker_threads_.size()); | |
| 333 | |
| 334 if (idle_worker_threads_stack_.Size() == worker_threads_.size()) | |
| 335 idle_worker_threads_stack_cv_for_testing_->Broadcast(); | |
| 336 } | |
| 337 | |
| 338 } // namespace internal | |
| 339 } // namespace base | |
| OLD | NEW |
|---|
Chromium Code Reviews
Issue 1906083002:
TaskScheduler: Remove base/task_scheduler/utils.h/.cc (Closed)
Base URL: https://chromium.googlesource.com/chromium/src.git@sched_2_stack