| 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 <algorithm> | 7 #include <algorithm> |
| 8 #include <utility> | 8 #include <utility> |
| 9 | 9 |
| 10 #include "base/bind.h" | 10 #include "base/bind.h" |
| (...skipping 258 matching lines...) Expand 10 before | Expand all | Expand 10 after Loading... |
| 269 } | 269 } |
| 270 } | 270 } |
| 271 | 271 |
| 272 NOTREACHED(); | 272 NOTREACHED(); |
| 273 return nullptr; | 273 return nullptr; |
| 274 } | 274 } |
| 275 | 275 |
| 276 void SchedulerThreadPoolImpl::ReEnqueueSequence( | 276 void SchedulerThreadPoolImpl::ReEnqueueSequence( |
| 277 scoped_refptr<Sequence> sequence, | 277 scoped_refptr<Sequence> sequence, |
| 278 const SequenceSortKey& sequence_sort_key) { | 278 const SequenceSortKey& sequence_sort_key) { |
| 279 shared_priority_queue_.BeginTransaction()->Push( | 279 shared_priority_queue_.BeginTransaction()->Push(std::move(sequence), |
| 280 WrapUnique(new PriorityQueue::SequenceAndSortKey(std::move(sequence), | 280 sequence_sort_key); |
| 281 sequence_sort_key))); | |
| 282 | 281 |
| 283 // The thread calling this method just ran a Task from |sequence| and will | 282 // The thread calling this method just ran a Task from |sequence| and will |
| 284 // soon try to get another Sequence from which to run a Task. If the thread | 283 // soon try to get another Sequence from which to run a Task. If the thread |
| 285 // belongs to this pool, it will get that Sequence from | 284 // belongs to this pool, it will get that Sequence from |
| 286 // |shared_priority_queue_|. When that's the case, there is no need to wake up | 285 // |shared_priority_queue_|. When that's the case, there is no need to wake up |
| 287 // another thread after |sequence| is inserted in |shared_priority_queue_|. If | 286 // another thread after |sequence| is inserted in |shared_priority_queue_|. If |
| 288 // we did wake up another thread, we would waste resources by having more | 287 // we did wake up another thread, we would waste resources by having more |
| 289 // threads trying to get a Sequence from |shared_priority_queue_| than the | 288 // threads trying to get a Sequence from |shared_priority_queue_| than the |
| 290 // number of Sequences in it. | 289 // number of Sequences in it. |
| 291 if (tls_current_thread_pool.Get().Get() != this) | 290 if (tls_current_thread_pool.Get().Get() != this) |
| (...skipping 48 matching lines...) Expand 10 before | Expand all | Expand 10 after Loading... |
| 340 | 339 |
| 341 const bool sequence_was_empty = sequence->PushTask(std::move(task)); | 340 const bool sequence_was_empty = sequence->PushTask(std::move(task)); |
| 342 if (sequence_was_empty) { | 341 if (sequence_was_empty) { |
| 343 // Insert |sequence| in |priority_queue| if it was empty before |task| was | 342 // Insert |sequence| in |priority_queue| if it was empty before |task| was |
| 344 // inserted into it. Otherwise, one of these must be true: | 343 // inserted into it. Otherwise, one of these must be true: |
| 345 // - |sequence| is already in a PriorityQueue (not necessarily | 344 // - |sequence| is already in a PriorityQueue (not necessarily |
| 346 // |shared_priority_queue_|), or, | 345 // |shared_priority_queue_|), or, |
| 347 // - A worker thread is running a Task from |sequence|. It will insert | 346 // - A worker thread is running a Task from |sequence|. It will insert |
| 348 // |sequence| in a PriorityQueue once it's done running the Task. | 347 // |sequence| in a PriorityQueue once it's done running the Task. |
| 349 const auto sequence_sort_key = sequence->GetSortKey(); | 348 const auto sequence_sort_key = sequence->GetSortKey(); |
| 350 priority_queue->BeginTransaction()->Push( | 349 priority_queue->BeginTransaction()->Push(std::move(sequence), |
| 351 WrapUnique(new PriorityQueue::SequenceAndSortKey(std::move(sequence), | 350 sequence_sort_key); |
| 352 sequence_sort_key))); | |
| 353 | 351 |
| 354 // Wake up a worker thread to process |sequence|. | 352 // Wake up a worker thread to process |sequence|. |
| 355 if (worker_thread) | 353 if (worker_thread) |
| 356 worker_thread->WakeUp(); | 354 worker_thread->WakeUp(); |
| 357 else | 355 else |
| 358 WakeUpOneThread(); | 356 WakeUpOneThread(); |
| 359 } | 357 } |
| 360 } | 358 } |
| 361 | 359 |
| 362 SchedulerThreadPoolImpl::SchedulerWorkerThreadDelegateImpl:: | 360 SchedulerThreadPoolImpl::SchedulerWorkerThreadDelegateImpl:: |
| (...skipping 25 matching lines...) Expand all Loading... |
| 388 | 386 |
| 389 scoped_refptr<Sequence> | 387 scoped_refptr<Sequence> |
| 390 SchedulerThreadPoolImpl::SchedulerWorkerThreadDelegateImpl::GetWork( | 388 SchedulerThreadPoolImpl::SchedulerWorkerThreadDelegateImpl::GetWork( |
| 391 SchedulerWorkerThread* worker_thread) { | 389 SchedulerWorkerThread* worker_thread) { |
| 392 DCHECK(ContainsWorkerThread(outer_->worker_threads_, worker_thread)); | 390 DCHECK(ContainsWorkerThread(outer_->worker_threads_, worker_thread)); |
| 393 | 391 |
| 394 scoped_refptr<Sequence> sequence; | 392 scoped_refptr<Sequence> sequence; |
| 395 { | 393 { |
| 396 std::unique_ptr<PriorityQueue::Transaction> shared_transaction( | 394 std::unique_ptr<PriorityQueue::Transaction> shared_transaction( |
| 397 outer_->shared_priority_queue_.BeginTransaction()); | 395 outer_->shared_priority_queue_.BeginTransaction()); |
| 398 const auto& shared_sequence_and_sort_key = shared_transaction->Peek(); | |
| 399 | |
| 400 std::unique_ptr<PriorityQueue::Transaction> single_threaded_transaction( | 396 std::unique_ptr<PriorityQueue::Transaction> single_threaded_transaction( |
| 401 single_threaded_priority_queue_.BeginTransaction()); | 397 single_threaded_priority_queue_.BeginTransaction()); |
| 402 const auto& single_threaded_sequence_and_sort_key = | |
| 403 single_threaded_transaction->Peek(); | |
| 404 | 398 |
| 405 if (shared_sequence_and_sort_key.is_null() && | 399 if (shared_transaction->IsEmpty() && |
| 406 single_threaded_sequence_and_sort_key.is_null()) { | 400 single_threaded_transaction->IsEmpty()) { |
| 407 single_threaded_transaction.reset(); | 401 single_threaded_transaction.reset(); |
| 408 | 402 |
| 409 // |shared_transaction| is kept alive while |worker_thread| is added to | 403 // |shared_transaction| is kept alive while |worker_thread| is added to |
| 410 // |idle_worker_threads_stack_| to avoid this race: | 404 // |idle_worker_threads_stack_| to avoid this race: |
| 411 // 1. This thread creates a Transaction, finds |shared_priority_queue_| | 405 // 1. This thread creates a Transaction, finds |shared_priority_queue_| |
| 412 // empty and ends the Transaction. | 406 // empty and ends the Transaction. |
| 413 // 2. Other thread creates a Transaction, inserts a Sequence into | 407 // 2. Other thread creates a Transaction, inserts a Sequence into |
| 414 // |shared_priority_queue_| and ends the Transaction. This can't happen | 408 // |shared_priority_queue_| and ends the Transaction. This can't happen |
| 415 // if the Transaction of step 1 is still active because because there | 409 // if the Transaction of step 1 is still active because because there |
| 416 // can only be one active Transaction per PriorityQueue at a time. | 410 // can only be one active Transaction per PriorityQueue at a time. |
| 417 // 3. Other thread calls WakeUpOneThread(). No thread is woken up because | 411 // 3. Other thread calls WakeUpOneThread(). No thread is woken up because |
| 418 // |idle_worker_threads_stack_| is empty. | 412 // |idle_worker_threads_stack_| is empty. |
| 419 // 4. This thread adds itself to |idle_worker_threads_stack_| and goes to | 413 // 4. This thread adds itself to |idle_worker_threads_stack_| and goes to |
| 420 // sleep. No thread runs the Sequence inserted in step 2. | 414 // sleep. No thread runs the Sequence inserted in step 2. |
| 421 outer_->AddToIdleWorkerThreadsStack(worker_thread); | 415 outer_->AddToIdleWorkerThreadsStack(worker_thread); |
| 422 return nullptr; | 416 return nullptr; |
| 423 } | 417 } |
| 424 | 418 |
| 425 // True if both PriorityQueues have Sequences and the Sequence at the top of | 419 // True if both PriorityQueues have Sequences and the Sequence at the top of |
| 426 // the shared PriorityQueue is more important. | 420 // the shared PriorityQueue is more important. |
| 427 const bool shared_sequence_is_more_important = | 421 const bool shared_sequence_is_more_important = |
| 428 !shared_sequence_and_sort_key.is_null() && | 422 !shared_transaction->IsEmpty() && |
| 429 !single_threaded_sequence_and_sort_key.is_null() && | 423 !single_threaded_transaction->IsEmpty() && |
| 430 shared_sequence_and_sort_key.sort_key > | 424 shared_transaction->PeekSortKey() > |
| 431 single_threaded_sequence_and_sort_key.sort_key; | 425 single_threaded_transaction->PeekSortKey(); |
| 432 | 426 |
| 433 if (single_threaded_sequence_and_sort_key.is_null() || | 427 if (single_threaded_transaction->IsEmpty() || |
| 434 shared_sequence_is_more_important) { | 428 shared_sequence_is_more_important) { |
| 435 sequence = shared_sequence_and_sort_key.sequence; | 429 sequence = shared_transaction->PopSequence(); |
| 436 shared_transaction->Pop(); | |
| 437 last_sequence_is_single_threaded_ = false; | 430 last_sequence_is_single_threaded_ = false; |
| 438 } else { | 431 } else { |
| 439 DCHECK(!single_threaded_sequence_and_sort_key.is_null()); | 432 DCHECK(!single_threaded_transaction->IsEmpty()); |
| 440 sequence = single_threaded_sequence_and_sort_key.sequence; | 433 sequence = single_threaded_transaction->PopSequence(); |
| 441 single_threaded_transaction->Pop(); | |
| 442 last_sequence_is_single_threaded_ = true; | 434 last_sequence_is_single_threaded_ = true; |
| 443 } | 435 } |
| 444 } | 436 } |
| 445 DCHECK(sequence); | 437 DCHECK(sequence); |
| 446 | 438 |
| 447 outer_->RemoveFromIdleWorkerThreadsStack(worker_thread); | 439 outer_->RemoveFromIdleWorkerThreadsStack(worker_thread); |
| 448 return sequence; | 440 return sequence; |
| 449 } | 441 } |
| 450 | 442 |
| 451 void SchedulerThreadPoolImpl::SchedulerWorkerThreadDelegateImpl:: | 443 void SchedulerThreadPoolImpl::SchedulerWorkerThreadDelegateImpl:: |
| 452 ReEnqueueSequence(scoped_refptr<Sequence> sequence) { | 444 ReEnqueueSequence(scoped_refptr<Sequence> sequence) { |
| 453 if (last_sequence_is_single_threaded_) { | 445 if (last_sequence_is_single_threaded_) { |
| 454 // A single-threaded Sequence is always re-enqueued in the single-threaded | 446 // A single-threaded Sequence is always re-enqueued in the single-threaded |
| 455 // PriorityQueue from which it was extracted. | 447 // PriorityQueue from which it was extracted. |
| 456 const SequenceSortKey sequence_sort_key = sequence->GetSortKey(); | 448 const SequenceSortKey sequence_sort_key = sequence->GetSortKey(); |
| 457 single_threaded_priority_queue_.BeginTransaction()->Push( | 449 single_threaded_priority_queue_.BeginTransaction()->Push( |
| 458 WrapUnique(new PriorityQueue::SequenceAndSortKey(std::move(sequence), | 450 std::move(sequence), sequence_sort_key); |
| 459 sequence_sort_key))); | |
| 460 } else { | 451 } else { |
| 461 // |re_enqueue_sequence_callback_| will determine in which PriorityQueue | 452 // |re_enqueue_sequence_callback_| will determine in which PriorityQueue |
| 462 // |sequence| must be enqueued. | 453 // |sequence| must be enqueued. |
| 463 re_enqueue_sequence_callback_.Run(std::move(sequence)); | 454 re_enqueue_sequence_callback_.Run(std::move(sequence)); |
| 464 } | 455 } |
| 465 } | 456 } |
| 466 | 457 |
| 467 SchedulerThreadPoolImpl::SchedulerThreadPoolImpl( | 458 SchedulerThreadPoolImpl::SchedulerThreadPoolImpl( |
| 468 TaskTracker* task_tracker, | 459 TaskTracker* task_tracker, |
| 469 DelayedTaskManager* delayed_task_manager) | 460 DelayedTaskManager* delayed_task_manager) |
| (...skipping 59 matching lines...) Expand 10 before | Expand all | Expand 10 after Loading... |
| 529 } | 520 } |
| 530 | 521 |
| 531 void SchedulerThreadPoolImpl::RemoveFromIdleWorkerThreadsStack( | 522 void SchedulerThreadPoolImpl::RemoveFromIdleWorkerThreadsStack( |
| 532 SchedulerWorkerThread* worker_thread) { | 523 SchedulerWorkerThread* worker_thread) { |
| 533 AutoSchedulerLock auto_lock(idle_worker_threads_stack_lock_); | 524 AutoSchedulerLock auto_lock(idle_worker_threads_stack_lock_); |
| 534 idle_worker_threads_stack_.Remove(worker_thread); | 525 idle_worker_threads_stack_.Remove(worker_thread); |
| 535 } | 526 } |
| 536 | 527 |
| 537 } // namespace internal | 528 } // namespace internal |
| 538 } // namespace base | 529 } // namespace base |
| OLD | NEW |