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