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 |