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