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..9294a66b8cd3511f3a2796b531dc3af504d47472 |
--- /dev/null |
+++ b/base/task_scheduler/worker_thread.cc |
@@ -0,0 +1,258 @@ |
+// 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 <ostream> |
+#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) { |
+ // Tasks are never nested on WorkerThread. |
+ return PostDelayedTask(from_here, task, delay); |
+} |
+ |
+SchedulerSingleThreadTaskRunner::~SchedulerSingleThreadTaskRunner() = default; |
+ |
+} // namespace |
+ |
+scoped_ptr<WorkerThread> WorkerThread::CreateWorkerThread( |
+ ThreadPriority thread_priority, |
+ PriorityQueue* shared_priority_queue, |
+ const SharedSequenceStillHasTasksCallback& |
+ shared_sequence_still_has_tasks_callback, |
+ const StateChangedCallback& state_changed_callback, |
+ TaskTracker* task_tracker) { |
+ scoped_ptr<WorkerThread> worker_thread( |
+ new WorkerThread(thread_priority, shared_priority_queue, |
+ shared_sequence_still_has_tasks_callback, |
+ state_changed_callback, task_tracker)); |
+ |
+ if (worker_thread->thread_handle_.is_null()) |
+ return scoped_ptr<WorkerThread>(); |
+ return worker_thread; |
+} |
+ |
+WorkerThread::~WorkerThread() { |
+ DCHECK(should_exit_for_testing()); |
+} |
+ |
+void WorkerThread::WakeUp() { |
+ wake_up_event_.Signal(); |
+} |
+ |
+scoped_refptr<SingleThreadTaskRunner> WorkerThread::CreateTaskRunnerWithTraits( |
+ const TaskTraits& traits) { |
+ // A WorkerThread is never destroyed, except in tests in which we don't use |
+ // task runners after their WorkerThread has been destroyed. Because of that, |
+ // it is correct to keep pointers to WorkerThread members in the constructed |
+ // task runner. |
+ return scoped_refptr<SingleThreadTaskRunner>( |
+ new SchedulerSingleThreadTaskRunner( |
+ traits, &single_thread_priority_queue_, task_tracker_)); |
+} |
+ |
+void WorkerThread::JoinForTesting() { |
+ should_exit_for_testing_ = true; |
+ base::subtle::MemoryBarrier(); |
+ WakeUp(); |
+ PlatformThread::Join(thread_handle_); |
+} |
+ |
+WorkerThread::WorkerThread(ThreadPriority thread_priority, |
+ PriorityQueue* shared_priority_queue, |
+ const SharedSequenceStillHasTasksCallback& |
+ shared_sequence_still_has_tasks_callback, |
+ const StateChangedCallback& state_changed_callback, |
+ TaskTracker* task_tracker) |
+ : wake_up_event_(false, false), |
+ single_thread_priority_queue_( |
+ Bind(&WorkerThread::WakeUp, Unretained(this)), |
+ shared_priority_queue), |
+ shared_priority_queue_(shared_priority_queue), |
+ shared_sequence_still_has_tasks_callback_( |
+ shared_sequence_still_has_tasks_callback), |
+ state_changed_callback_(state_changed_callback), |
+ task_tracker_(task_tracker) { |
+ DCHECK(shared_priority_queue_); |
+ DCHECK(!shared_sequence_still_has_tasks_callback_.is_null()); |
+ DCHECK(!state_changed_callback_.is_null()); |
+ DCHECK(task_tracker_); |
+ |
+ const size_t kDefaultStackSize = 0; |
robliao
2016/03/24 23:03:26
Might as well make this static const.
fdoray
2016/03/29 18:33:34
Done.
|
+ PlatformThread::CreateWithPriority(kDefaultStackSize, this, &thread_handle_, |
+ thread_priority); |
+} |
+ |
+scoped_refptr<Sequence> WorkerThread::GetWork(bool* is_single_threaded) { |
+ DCHECK(is_single_threaded); |
robliao
2016/03/24 23:03:26
I think this is one of the rare cases where a DCHE
fdoray
2016/03/29 18:33:34
Done.
|
+ *is_single_threaded = 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(); |
+ *is_single_threaded = true; |
+ return single_thread_sequence.sequence; |
+} |
+ |
+void WorkerThread::SetState(State state) { |
+ DCHECK_NE(state_, state); |
+ state_ = state; |
+ state_changed_callback_.Run(this, state); |
+} |
+ |
+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; |
+ scoped_refptr<Sequence> sequence = GetWork(&sequence_is_single_threaded); |
+ if (!sequence) { |
+ SetState(State::IDLE); |
+ |
+ // Check one more time if there is work available. Work could have been |
+ // added to |shared_priority_queue_| between the first call to GetWork() |
+ // and the call to SetState() above in which case this WorkerThread |
+ // shouldn't go to sleep (the recipient of |state_changed_callback_| |
+ // hadn't be notified that this WorkerThread was idle and might have |
robliao
2016/03/24 23:03:26
Nit: hasn't
fdoray
2016/03/29 18:33:34
I don't think that "hasn't" is correct because at
robliao
2016/03/30 00:42:42
This CL comment is no longer relevant as of patchs
|
+ // assumed that it was gonna run the newly added work). Note that |
+ // |wake_up_event_| is always signaled when work is added to |
+ // |single_thread_priority_queue_|. |
+ sequence = GetWork(&sequence_is_single_threaded); |
+ |
+ if (!sequence) { |
+ wake_up_event_.Wait(); |
+ |
+ // Always mark this WorkerThread as BUSY after a wake up, regardless of |
+ // whether work is actually found in its PriorityQueues. This ensures |
+ // that there won't be 2 consecutive IDLE invocations of |
+ // |state_changed_callback_| in case of spurious wake up. |
+ SetState(State::BUSY); |
+ |
+ continue; |
+ } |
+ |
+ SetState(State::BUSY); |
+ wake_up_event_.Reset(); |
robliao
2016/03/24 23:03:26
Why is this reset necessary since our events are c
fdoray
2016/03/29 18:33:34
I moved the reset at the bottom of this method and
|
+ } |
+ |
+ DCHECK_EQ(state_, State::BUSY); |
+ |
+ // Peek the next task in |sequence| and run it. |
+ task_tracker_->RunTask(sequence->PeekTask()); |
+ |
+ // Pop a task from |sequence|. If it is empty after this, reinsert it in |
robliao
2016/03/24 23:03:26
I think you mean if the sequence isn't empty.
I w
fdoray
2016/03/29 18:33:34
We don't necessary reinsert the sequence in its or
|
+ // |single_thread_priority_queue_| or notify |
+ // |shared_sequence_still_has_tasks_callback_|. |
+ if (!sequence->PopTask()) { |
+ if (sequence_is_single_threaded) { |
+ const SequenceSortKey sort_key = sequence->GetSortKey(); |
+ single_thread_priority_queue_.BeginTransaction()->Push( |
+ make_scoped_ptr(new PriorityQueue::SequenceAndSortKey( |
+ std::move(sequence), sort_key))); |
+ } else { |
+ shared_sequence_still_has_tasks_callback_.Run(this, |
+ std::move(sequence)); |
+ } |
+ } |
+ } |
+} |
+ |
+std::ostream& operator<<(std::ostream& os, WorkerThread::State state) { |
+ switch (state) { |
+ case WorkerThread::State::BUSY: |
+ os << "BUSY"; |
+ break; |
+ case WorkerThread::State::IDLE: |
+ os << "IDLE"; |
+ break; |
+ } |
+ return os; |
+} |
+ |
+} // namespace internal |
+} // namespace base |