Index: base/task_scheduler/worker_thread.cc |
diff --git a/base/task_scheduler/worker_thread.cc b/base/task_scheduler/worker_thread.cc |
new file mode 100644 |
index 0000000000000000000000000000000000000000..27f919168cb2205ea74cf21cde33f71403bb3f44 |
--- /dev/null |
+++ b/base/task_scheduler/worker_thread.cc |
@@ -0,0 +1,251 @@ |
+// Copyright 2016 The Chromium Authors. All rights reserved. |
+// Use of this source code is governed by a BSD-style license that can be |
+// found in the LICENSE file. |
+ |
+#include "base/task_scheduler/worker_thread.h" |
+ |
+#include <utility> |
+ |
+#include "base/bind.h" |
+#include "base/logging.h" |
+#include "base/task_scheduler/task_tracker.h" |
+#include "base/task_scheduler/utils.h" |
+#include "base/time/time.h" |
+ |
+namespace base { |
+namespace internal { |
+ |
+namespace { |
+ |
+class SchedulerSingleThreadTaskRunner : public SingleThreadTaskRunner { |
+ public: |
+ SchedulerSingleThreadTaskRunner(const TaskTraits& traits, |
+ PriorityQueue* single_thread_priority_queue, |
+ TaskTracker* task_tracker); |
+ |
+ // SingleThreadTaskRunner: |
+ bool PostNonNestableDelayedTask(const tracked_objects::Location& from_here, |
+ const Closure& task, |
+ TimeDelta delay) override; |
+ bool PostDelayedTask(const tracked_objects::Location& from_here, |
+ const Closure& closure, |
+ TimeDelta delay) override; |
+ bool RunsTasksOnCurrentThread() const override; |
+ |
+ private: |
+ ~SchedulerSingleThreadTaskRunner() override; |
+ |
+ const TaskTraits traits_; |
+ const scoped_refptr<Sequence> sequence_; |
+ PriorityQueue* const priority_queue_; |
+ TaskTracker* const task_tracker_; |
+ |
+ DISALLOW_COPY_AND_ASSIGN(SchedulerSingleThreadTaskRunner); |
+}; |
+ |
+SchedulerSingleThreadTaskRunner::SchedulerSingleThreadTaskRunner( |
+ const TaskTraits& traits, |
+ PriorityQueue* single_thread_priority_queue, |
+ TaskTracker* task_tracker) |
+ : traits_(traits), |
+ sequence_(new Sequence), |
+ priority_queue_(single_thread_priority_queue), |
+ task_tracker_(task_tracker) {} |
+ |
+bool SchedulerSingleThreadTaskRunner::PostDelayedTask( |
+ const tracked_objects::Location& from_here, |
+ const Closure& closure, |
+ TimeDelta delay) { |
+ // TODO(fdoray): Support delayed tasks. |
+ DCHECK(delay.is_zero()); |
+ PostTaskHelper(make_scoped_ptr(new Task(from_here, closure, traits_)), |
+ sequence_, priority_queue_, task_tracker_); |
+ return true; |
+} |
+ |
+bool SchedulerSingleThreadTaskRunner::RunsTasksOnCurrentThread() const { |
+ // TODO(fdoray): Return true only if tasks posted may actually run on the |
+ // current thread. It is valid, but not ideal, to always return true. |
+ return true; |
+} |
+ |
+bool SchedulerSingleThreadTaskRunner::PostNonNestableDelayedTask( |
+ const tracked_objects::Location& from_here, |
+ const Closure& task, |
+ TimeDelta delay) { |
+ return PostDelayedTask(from_here, task, delay); |
+} |
+ |
+SchedulerSingleThreadTaskRunner::~SchedulerSingleThreadTaskRunner() = default; |
+ |
+void PushSequenceInPriorityQueue(scoped_refptr<Sequence> sequence, |
+ PriorityQueue* priority_queue) { |
+ const SequenceSortKey sort_key = sequence->GetSortKey(); |
+ priority_queue->BeginTransaction()->Push(make_scoped_ptr( |
+ new PriorityQueue::SequenceAndSortKey(std::move(sequence), sort_key))); |
+} |
+ |
+} // namespace |
+ |
+scoped_ptr<WorkerThread> WorkerThread::CreateWorkerThread( |
+ ThreadPriority thread_priority, |
+ PriorityQueue* shared_priority_queue, |
+ const ReinsertSequenceCallback& reinsert_sequence_callback, |
+ const BecomesIdleCallback& becomes_idle_callback, |
+ TaskTracker* task_tracker) { |
+ scoped_ptr<WorkerThread> worker_thread(new WorkerThread( |
+ thread_priority, shared_priority_queue, reinsert_sequence_callback, |
+ becomes_idle_callback, task_tracker)); |
+ |
+ if (worker_thread->thread_handle_.is_null()) |
+ return scoped_ptr<WorkerThread>(); |
+ return worker_thread; |
+} |
+ |
+WorkerThread::~WorkerThread() { |
+ AutoSchedulerLock auto_lock(lock_); |
+ DCHECK(should_exit_for_testing_); |
+} |
+ |
+bool WorkerThread::WakeUp() { |
+ AutoSchedulerLock auto_lock(lock_); |
+ if (is_awake_) |
+ return false; |
+ is_awake_ = true; |
+ wake_up_cv_->Signal(); |
+ return true; |
+} |
+ |
+scoped_refptr<SingleThreadTaskRunner> WorkerThread::CreateTaskRunnerWithTraits( |
+ const TaskTraits& traits) { |
+ return scoped_refptr<SingleThreadTaskRunner>( |
+ new SchedulerSingleThreadTaskRunner( |
+ traits, &single_thread_priority_queue_, task_tracker_)); |
+} |
+ |
+void WorkerThread::JoinForTesting() { |
+ { |
+ AutoSchedulerLock auto_lock(lock_); |
+ DCHECK(!should_exit_for_testing_); |
+ should_exit_for_testing_ = true; |
+ } |
+ WakeUp(); |
+ PlatformThread::Join(thread_handle_); |
+} |
+ |
+WorkerThread::WorkerThread( |
+ ThreadPriority thread_priority, |
+ PriorityQueue* shared_priority_queue, |
+ const ReinsertSequenceCallback& reinsert_sequence_callback, |
+ const BecomesIdleCallback& becomes_idle_callback, |
+ TaskTracker* task_tracker) |
+ : wake_up_cv_(lock_.CreateConditionVariable()), |
+ is_awake_(true), |
+ should_exit_for_testing_(false), |
+ single_thread_priority_queue_( |
+ Bind(IgnoreResult(&WorkerThread::WakeUp), Unretained(this)), |
+ shared_priority_queue), |
+ shared_priority_queue_(shared_priority_queue), |
+ reinsert_sequence_callback_(reinsert_sequence_callback), |
+ becomes_idle_callback_(becomes_idle_callback), |
+ task_tracker_(task_tracker) { |
+ DCHECK(shared_priority_queue_); |
+ DCHECK(!reinsert_sequence_callback_.is_null()); |
+ DCHECK(!becomes_idle_callback_.is_null()); |
+ DCHECK(task_tracker_); |
+ |
+#if defined(OS_MACOSX) |
+ // Mac only supports 2 priorities. crbug.com/554651 |
+ if (thread_priority != ThreadPriority::NORMAL && |
+ thread_priority != ThreadPriority::REALTIME_AUDIO) { |
+ thread_priority = ThreadPriority::NORMAL; |
+ } |
+#endif // defined(OS_MACOSX) |
+ |
+ const size_t kDefaultStackSize = 0; |
+ PlatformThread::CreateWithPriority(kDefaultStackSize, this, &thread_handle_, |
+ thread_priority); |
+} |
+ |
+scoped_refptr<Sequence> WorkerThread::GetWork(bool* single_thread) { |
+ DCHECK(single_thread); |
+ *single_thread = false; |
+ |
+ scoped_ptr<PriorityQueue::Transaction> shared_transaction( |
+ shared_priority_queue_->BeginTransaction()); |
+ const PriorityQueue::SequenceAndSortKey shared_sequence = |
+ shared_transaction->Peek(); |
+ |
+ scoped_ptr<PriorityQueue::Transaction> single_thread_transaction( |
+ single_thread_priority_queue_.BeginTransaction()); |
+ const PriorityQueue::SequenceAndSortKey single_thread_sequence = |
+ single_thread_transaction->Peek(); |
+ |
+ if (single_thread_sequence.is_null() && shared_sequence.is_null()) { |
+ return scoped_refptr<Sequence>(); |
+ } |
+ |
+ if (single_thread_sequence.is_null() || |
+ (!shared_sequence.is_null() && |
+ single_thread_sequence.sort_key < shared_sequence.sort_key)) { |
+ shared_transaction->Pop(); |
+ return shared_sequence.sequence; |
+ } |
+ |
+ DCHECK(!single_thread_sequence.is_null()); |
+ single_thread_transaction->Pop(); |
+ *single_thread = true; |
+ return single_thread_sequence.sequence; |
+} |
+ |
+void WorkerThread::WaitUntilWakeUp() { |
+ AutoSchedulerLock auto_lock(lock_); |
+ while (!is_awake_) |
+ wake_up_cv_->Wait(); |
+} |
+ |
+void WorkerThread::ThreadMain() { |
+ while (!task_tracker_->shutdown_completed() && !should_exit_for_testing_) { |
+ // Get the sequence containing the next task to execute. |
+ bool sequence_is_single_threaded = false; |
robliao
2016/03/09 15:41:12
For unit tests:
If we're exiting the test, call Wa
fdoray
2016/03/14 20:46:35
You're right. Thanks for the fix. I moved it insid
|
+ scoped_refptr<Sequence> sequence = GetWork(&sequence_is_single_threaded); |
+ if (sequence.get() == nullptr) { |
+ // Mark the WorkerThread as idle. |
+ { |
+ AutoSchedulerLock auto_lock(lock_); |
+ is_awake_ = false; |
+ } |
+ becomes_idle_callback_.Run(this); |
+ |
+ // Check one more time if there is work available. If work has been added |
+ // to |shared_priority_queue_| after the first call to GetWork() but |
+ // before the |becomes_idle_callback_| invocation, this WorkerThread |
+ // should run this work. |
+ sequence = GetWork(&sequence_is_single_threaded); |
+ |
+ if (sequence.get() == nullptr) { |
+ WaitUntilWakeUp(); |
+ sequence = GetWork(&sequence_is_single_threaded); |
+ } |
+ } |
+ |
+ if (sequence.get() != nullptr) { |
+ // Peek the next task in the sequence and run it. |
+ task_tracker_->RunTask(sequence->PeekTask()); |
+ |
+ // Pop the task from its sequence. If the sequence isn't empty, reinsert |
+ // it in the appropriate PriorityQueue. |
+ if (!sequence->PopTask()) { |
+ if (sequence_is_single_threaded) { |
+ PushSequenceInPriorityQueue(std::move(sequence), |
+ &single_thread_priority_queue_); |
+ } else { |
+ reinsert_sequence_callback_.Run(std::move(sequence), this); |
+ } |
+ } |
+ } |
+ } |
+} |
+ |
+} // namespace internal |
+} // namespace base |