Chromium Code Reviews
chromiumcodereview-hr@appspot.gserviceaccount.com (chromiumcodereview-hr) | Please choose your nickname with Settings | Help | Chromium Project | Gerrit Changes | Sign out
(2982)

Unified Diff: base/task_scheduler/worker_thread.cc

Issue 1704113002: TaskScheduler [6] SchedulerWorkerThread (Closed) Base URL: https://chromium.googlesource.com/chromium/src.git@s_4_shutdown
Patch Set: rebase Created 4 years, 9 months ago
Use n/p to move between diff chunks; N/P to move between comments. Draft comments are only viewable by you.
Jump to:
View side-by-side diff with in-line comments
Download patch
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

Powered by Google App Engine
This is Rietveld 408576698