Index: base/task_scheduler/worker_thread_unittest.cc |
diff --git a/base/task_scheduler/worker_thread_unittest.cc b/base/task_scheduler/worker_thread_unittest.cc |
new file mode 100644 |
index 0000000000000000000000000000000000000000..a61d1980c468bb91d9f7f9f92bf61b89cdaed55c |
--- /dev/null |
+++ b/base/task_scheduler/worker_thread_unittest.cc |
@@ -0,0 +1,457 @@ |
+// 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 <vector> |
+ |
+#include "base/bind.h" |
+#include "base/bind_helpers.h" |
+#include "base/callback_forward.h" |
+#include "base/logging.h" |
+#include "base/macros.h" |
+#include "base/memory/scoped_ptr.h" |
+#include "base/synchronization/condition_variable.h" |
+#include "base/task_scheduler/priority_queue.h" |
+#include "base/task_scheduler/scheduler_lock.h" |
+#include "base/task_scheduler/task_tracker.h" |
+#include "base/task_scheduler/utils.h" |
+#include "base/threading/simple_thread.h" |
+#include "testing/gtest/include/gtest/gtest.h" |
+ |
+namespace base { |
+namespace internal { |
+ |
+namespace { |
+ |
+const size_t kNumTasksPerTest = 500; |
robliao
2016/03/24 23:03:27
Why 500?
fdoray
2016/03/29 18:33:35
A high number of tasks increases the chances to ca
|
+ |
+class TaskClosureFactory { |
robliao
2016/03/24 23:03:26
Optional: Maybe SequencedTaskClosureFactory?
fdoray
2016/03/29 18:33:34
Now I use the factory to create non-sequenced task
|
+ public: |
+ TaskClosureFactory() : run_task_cv_(lock_.CreateConditionVariable()) {} |
+ |
+ ~TaskClosureFactory() { |
+ AutoSchedulerLock auto_lock(lock_); |
+ EXPECT_EQ(num_factored_tasks_, num_run_tasks_); |
+ } |
+ |
+ Closure CreateTaskClosure() { |
+ AutoSchedulerLock auto_lock(lock_); |
+ ++num_factored_tasks_; |
+ return Bind(&TaskClosureFactory::RunTask, Unretained(this), |
+ num_factored_tasks_); |
+ } |
+ |
+ void WaitUntilLastFactoredTaskRan() { |
robliao
2016/03/24 23:03:26
WaitForAllTasksToRun
fdoray
2016/03/29 18:33:35
Done.
|
+ AutoSchedulerLock auto_lock(lock_); |
+ while (num_factored_tasks_ != num_run_tasks_) |
robliao
2016/03/24 23:03:26
Nit: num_run_tasks_ < num_created_tasks_
Makes it
fdoray
2016/03/29 18:33:35
Done.
|
+ run_task_cv_->Wait(); |
+ } |
+ |
+ private: |
+ void RunTask(size_t task_index) { |
+ AutoSchedulerLock auto_lock(lock_); |
+ |
+ if (task_index != num_run_tasks_ + 1) |
robliao
2016/03/24 23:03:27
If you allow task_index to be zero, you can avoid
fdoray
2016/03/29 18:33:34
Done.
|
+ ADD_FAILURE() << "Unexpected task execution order."; |
+ |
+ ++num_run_tasks_; |
+ run_task_cv_->Signal(); |
+ } |
+ |
+ // Synchronizes access to all members. |
+ SchedulerLock lock_; |
+ |
+ // Signaled when a task runs. |
+ scoped_ptr<ConditionVariable> run_task_cv_; |
+ |
+ // Number of times that CreateTaskClosure() has been called. |
+ size_t num_factored_tasks_ = 0; |
robliao
2016/03/24 23:03:26
num_created_tasks_
fdoray
2016/03/29 18:33:34
Done.
|
+ |
+ // Number of times that RunTask() has been called. |
+ size_t num_run_tasks_ = 0; |
+ |
+ DISALLOW_COPY_AND_ASSIGN(TaskClosureFactory); |
+}; |
+ |
+class ThreadPostingTasks : public SimpleThread { |
+ public: |
+ ThreadPostingTasks(WorkerThread* worker_thread) |
+ : SimpleThread("ThreadPostingTasks"), worker_thread_(worker_thread) {} |
+ |
+ void WaitUntilLastFactoredTaskRan() { |
+ factory_.WaitUntilLastFactoredTaskRan(); |
+ } |
+ |
+ private: |
+ void Run() override { |
+ auto task_runner = worker_thread_->CreateTaskRunnerWithTraits(TaskTraits()); |
+ |
+ for (size_t i = 0; i < kNumTasksPerTest; ++i) |
+ task_runner->PostTask(FROM_HERE, factory_.CreateTaskClosure()); |
+ } |
+ |
+ WorkerThread* const worker_thread_; |
+ TaskClosureFactory factory_; |
+ |
+ DISALLOW_COPY_AND_ASSIGN(ThreadPostingTasks); |
+}; |
+ |
+} // namespace |
+ |
+class TaskSchedulerWorkerThreadTest : public testing::Test { |
+ protected: |
+ TaskSchedulerWorkerThreadTest() |
+ : shared_priority_queue_(Bind(&DoNothing)), |
+ state_changed_callback_cv_(lock_.CreateConditionVariable()) {} |
+ |
+ virtual void SetUp() override { |
robliao
2016/03/24 23:03:26
Nit: virtual not needed with override.
Here and be
fdoray
2016/03/29 18:33:35
Done.
|
+ worker_thread_ = WorkerThread::CreateWorkerThread( |
+ ThreadPriority::NORMAL, &shared_priority_queue_, |
+ Bind(&TaskSchedulerWorkerThreadTest:: |
+ PoppedTaskFromSharedSequenceCallback, |
+ Unretained(this)), |
+ Bind(&TaskSchedulerWorkerThreadTest::StateChangedCallback, |
+ Unretained(this)), |
+ &task_tracker_); |
+ ASSERT_TRUE(worker_thread_); |
+ WaitUntilIdle(); |
+ |
+ AutoSchedulerLock auto_lock(lock_); |
+ num_state_changes_ = 0; |
+ } |
+ |
+ virtual void TearDown() override { worker_thread_->JoinForTesting(); } |
+ |
+ void WaitUntilIdle() { |
+ AutoSchedulerLock auto_lock(lock_); |
+ while (last_state_ != WorkerThread::State::IDLE) |
+ state_changed_callback_cv_->Wait(); |
+ } |
+ |
+ void WaitUntilNumStateChanges(size_t expected_num_state_changes) { |
+ AutoSchedulerLock auto_lock(lock_); |
+ while (num_state_changes_ != expected_num_state_changes) |
robliao
2016/03/24 23:03:27
Nit: num_state_changes_ < expected_num_state_chang
fdoray
2016/03/29 18:33:35
Done.
|
+ state_changed_callback_cv_->Wait(); |
+ } |
+ |
+ // Returns the number of state changes that occurred after SetUp() completed |
+ // its execution. |
+ size_t num_state_changes() const { |
+ AutoSchedulerLock auto_lock(lock_); |
+ return num_state_changes_; |
+ } |
+ |
+ size_t num_popped_task_from_shared_sequence() const { |
+ AutoSchedulerLock auto_lock(lock_); |
+ return num_popped_task_from_shared_sequence_; |
+ } |
+ |
+ PriorityQueue shared_priority_queue_; |
+ TaskTracker task_tracker_; |
+ scoped_ptr<WorkerThread> worker_thread_; |
+ |
+ private: |
+ void PoppedTaskFromSharedSequenceCallback(const WorkerThread* worker_thread, |
+ scoped_refptr<Sequence> sequence) { |
+ { |
+ AutoSchedulerLock auto_lock(lock_); |
+ ++num_popped_task_from_shared_sequence_; |
+ } |
+ |
+ // Reinsert sequence in |shared_priority_queue_|. |
+ const SequenceSortKey sort_key = sequence->GetSortKey(); |
+ shared_priority_queue_.BeginTransaction()->Push(make_scoped_ptr( |
+ new PriorityQueue::SequenceAndSortKey(std::move(sequence), sort_key))); |
+ } |
+ |
+ void StateChangedCallback(WorkerThread* worker_thread, |
+ WorkerThread::State state) { |
+ AutoSchedulerLock auto_lock(lock_); |
+ EXPECT_EQ(worker_thread_.get(), worker_thread); |
+ EXPECT_NE(last_state_, state); |
+ last_state_ = state; |
+ ++num_state_changes_; |
+ state_changed_callback_cv_->Signal(); |
+ } |
+ |
+ // Synchronizes access to all members below. |
+ mutable SchedulerLock lock_; |
+ |
+ // Condition variable signaled when StateChangedCallback() is invoked. |
+ scoped_ptr<ConditionVariable> state_changed_callback_cv_; |
+ |
+ // Last state reported to StateChangedCallback(). |
+ WorkerThread::State last_state_ = WorkerThread::State::BUSY; |
+ |
+ // Number of times that StateChangedCallback() has been invoked. |
+ size_t num_state_changes_ = 0; |
+ |
+ // Number of times that PoppedTaskFromSharedSequenceCallback() has been |
+ // invoked. |
+ size_t num_popped_task_from_shared_sequence_ = 0; |
+ |
+ DISALLOW_COPY_AND_ASSIGN(TaskSchedulerWorkerThreadTest); |
+}; |
+ |
+// Verify each call to WakeUp() on an idle WorkerThread causes 2 state changes. |
robliao
2016/03/24 23:03:26
This seems like a brittle test, especially if for
fdoray
2016/03/29 18:33:34
StateChangedCallback() verifies that the states go
|
+TEST_F(TaskSchedulerWorkerThreadTest, WakeUp) { |
+ for (size_t i = 0; i < kNumTasksPerTest; ++i) { |
+ worker_thread_->WakeUp(); |
+ WaitUntilNumStateChanges(2 * (i + 1)); |
robliao
2016/03/24 23:03:26
Should this instead WaitUntilIdle? If so, you can
fdoray
2016/03/29 18:33:34
Replacing WaitUntilNumStateChanges() with WaitUnti
|
+ EXPECT_EQ(2 * (i + 1), num_state_changes()); |
+ } |
+ |
+ EXPECT_EQ(0, num_popped_task_from_shared_sequence()); |
+} |
+ |
+// Verify that a task runs successfully when it is posted through a |
+// task runner returned by WorkerThread::CreateTaskRunnerWithTraits(). |
+TEST_F(TaskSchedulerWorkerThreadTest, PostOneSingleThreadedTask) { |
+ TaskClosureFactory factory; |
+ worker_thread_->CreateTaskRunnerWithTraits(TaskTraits()) |
+ ->PostTask(FROM_HERE, factory.CreateTaskClosure()); |
+ |
+ factory.WaitUntilLastFactoredTaskRan(); |
+ WaitUntilIdle(); |
+ EXPECT_EQ(2, num_state_changes()); |
+ EXPECT_EQ(0, num_popped_task_from_shared_sequence()); |
+} |
+ |
+// Verify that 500 tasks run successfully when they are posted through a task |
robliao
2016/03/24 23:03:27
Maybe |kNumTasksPerTest| instead of 500.
fdoray
2016/03/29 18:33:34
Done.
|
+// runner returned by WorkerThread::CreateTaskRunnerWithTraits(). Don't wait |
+// between posts. |
+TEST_F(TaskSchedulerWorkerThreadTest, |
+ PostMultipleSingleThreadedTasksNoWaitBetweenPosts) { |
+ TaskClosureFactory factory; |
+ auto task_runner = worker_thread_->CreateTaskRunnerWithTraits(TaskTraits()); |
+ |
+ for (size_t i = 0; i < kNumTasksPerTest; ++i) |
+ task_runner->PostTask(FROM_HERE, factory.CreateTaskClosure()); |
+ |
+ factory.WaitUntilLastFactoredTaskRan(); |
+ WaitUntilIdle(); |
+ EXPECT_GE(num_state_changes(), 2); |
+ EXPECT_EQ(0, num_popped_task_from_shared_sequence()); |
+} |
+ |
+// Verify that 500 tasks run successfully when they are posted through a task |
+// runner returned by WorkerThread::CreateTaskRunnerWithTraits(). Wait until the |
+// previous task has completed its execution before posting a new task. |
+TEST_F(TaskSchedulerWorkerThreadTest, |
robliao
2016/03/24 23:03:27
Isn't this similar to PostOneSingleThreadedTask, j
fdoray
2016/03/29 18:33:34
Yes, PostOneSingleThreadedTask is equivalent to th
|
+ PostMultipleSingleThreadedTasksWaitBetweenPosts) { |
+ TaskClosureFactory factory; |
+ auto task_runner = worker_thread_->CreateTaskRunnerWithTraits(TaskTraits()); |
+ |
+ for (size_t i = 0; i < kNumTasksPerTest; ++i) { |
+ task_runner->PostTask(FROM_HERE, factory.CreateTaskClosure()); |
+ factory.WaitUntilLastFactoredTaskRan(); |
+ WaitUntilIdle(); |
+ EXPECT_EQ(2 * (i + 1), num_state_changes()); |
+ } |
+ |
+ EXPECT_EQ(0, num_popped_task_from_shared_sequence()); |
+} |
+ |
+// Verify that 1000 tasks run successfully when they are posted through 2 task |
+// runners returned by WorkerThread::CreateTaskRunnerWithTraits(). Don't wait |
+// between posts. |
+TEST_F(TaskSchedulerWorkerThreadTest, |
+ PostMultipleTasksTwoSingleThreadedTaskRunners) { |
+ TaskClosureFactory factory; |
+ auto task_runner_a = worker_thread_->CreateTaskRunnerWithTraits(TaskTraits()); |
+ auto task_runner_b = worker_thread_->CreateTaskRunnerWithTraits(TaskTraits()); |
+ |
+ for (size_t i = 0; i < kNumTasksPerTest; ++i) { |
+ task_runner_a->PostTask(FROM_HERE, factory.CreateTaskClosure()); |
+ task_runner_b->PostTask(FROM_HERE, factory.CreateTaskClosure()); |
robliao
2016/03/24 23:03:26
Do we guarantee task execution ordering across tas
fdoray
2016/03/29 18:33:35
The current implementation runs tasks posted to th
|
+ } |
+ |
+ factory.WaitUntilLastFactoredTaskRan(); |
+ WaitUntilIdle(); |
+ EXPECT_GE(num_state_changes(), 2); |
+ EXPECT_EQ(0, num_popped_task_from_shared_sequence()); |
+} |
+ |
+// Verify that a task runs successfully when it is added to the shared priority |
+// queue of a WorkerThread. The test wakes up the WorkerThread after adding the |
+// task to the priority queue. |
robliao
2016/03/24 23:03:27
Mention that the wakeup is necessary since shared
fdoray
2016/03/29 18:33:35
Done.
|
+TEST_F(TaskSchedulerWorkerThreadTest, PostOneSharedTask) { |
+ TaskClosureFactory factory; |
+ PostTaskHelper(make_scoped_ptr(new Task( |
+ FROM_HERE, factory.CreateTaskClosure(), TaskTraits())), |
+ make_scoped_refptr(new Sequence), &shared_priority_queue_, |
+ &task_tracker_); |
+ |
+ worker_thread_->WakeUp(); |
+ factory.WaitUntilLastFactoredTaskRan(); |
+ WaitUntilIdle(); |
+ EXPECT_EQ(2, num_state_changes()); |
+ EXPECT_EQ(0, num_popped_task_from_shared_sequence()); |
+} |
+ |
+// Verify that 500 tasks run successfully when they are added to the shared |
+// priority queue of a WorkerThread in different Sequences. The test wakes up |
+// the WorkerThread each time it adds a task to the priority queue. |
robliao
2016/03/24 23:03:27
The worker thread looks like it's only woken up on
fdoray
2016/03/29 18:33:34
Done.
|
+TEST_F(TaskSchedulerWorkerThreadTest, |
+ PostMultipleSharedTasksNoWaitBetweenPosts) { |
+ TaskClosureFactory factory; |
+ for (size_t i = 0; i < kNumTasksPerTest; ++i) { |
+ PostTaskHelper(make_scoped_ptr(new Task( |
+ FROM_HERE, factory.CreateTaskClosure(), TaskTraits())), |
+ make_scoped_refptr(new Sequence), &shared_priority_queue_, |
+ &task_tracker_); |
+ } |
+ |
+ worker_thread_->WakeUp(); |
+ factory.WaitUntilLastFactoredTaskRan(); |
+ WaitUntilIdle(); |
+ EXPECT_EQ(2, num_state_changes()); |
+ EXPECT_EQ(0, num_popped_task_from_shared_sequence()); |
+} |
+ |
+// Verify that 500 tasks run successfully when they are added to the shared |
+// priority queue of a WorkerThread in different Sequences. The test wakes up |
+// the WorkerThread and waits until the task completes its execution each time |
+// it adds a task to the priority queue. |
+TEST_F(TaskSchedulerWorkerThreadTest, PostMultipleSharedTasksWaitBetweenPosts) { |
+ TaskClosureFactory factory; |
+ for (size_t i = 0; i < kNumTasksPerTest; ++i) { |
+ PostTaskHelper(make_scoped_ptr(new Task( |
+ FROM_HERE, factory.CreateTaskClosure(), TaskTraits())), |
+ make_scoped_refptr(new Sequence), &shared_priority_queue_, |
+ &task_tracker_); |
+ worker_thread_->WakeUp(); |
+ |
+ factory.WaitUntilLastFactoredTaskRan(); |
+ WaitUntilIdle(); |
+ EXPECT_EQ(2 * (i + 1), num_state_changes()); |
+ } |
+ |
+ EXPECT_EQ(0, num_popped_task_from_shared_sequence()); |
+} |
+ |
+// Verify that 500 tasks run successfully when they are added to the shared |
+// priority queue of a WorkerThread (all tasks in the same Sequence). The test |
+// wakes up the WorkerThread once it has added the 500 tasks to the priority |
+// queue. |
+TEST_F(TaskSchedulerWorkerThreadTest, PostMultipleSharedTasksInSameSequence) { |
+ TaskClosureFactory factory; |
+ scoped_refptr<Sequence> sequence(new Sequence); |
+ |
+ for (size_t i = 0; i < kNumTasksPerTest; ++i) { |
+ PostTaskHelper(make_scoped_ptr(new Task( |
+ FROM_HERE, factory.CreateTaskClosure(), TaskTraits())), |
+ sequence, &shared_priority_queue_, &task_tracker_); |
+ } |
+ |
+ worker_thread_->WakeUp(); |
+ factory.WaitUntilLastFactoredTaskRan(); |
+ WaitUntilIdle(); |
+ EXPECT_EQ(2, num_state_changes()); |
+ EXPECT_EQ(kNumTasksPerTest - 1, num_popped_task_from_shared_sequence()); |
+} |
+ |
+// Verify that 1000 tasks run successfully when they are added to the shared |
+// priority queue of a WorkerThread. Tasks are split into 2 sequences. The test |
+// wakes up the WorkerThread once it has added the 1000 tasks to the priority |
+// queue. |
+TEST_F(TaskSchedulerWorkerThreadTest, PostMultipleSharedTasksInTwoSequences) { |
+ TaskClosureFactory factory; |
+ scoped_refptr<Sequence> sequence_a(new Sequence); |
+ scoped_refptr<Sequence> sequence_b(new Sequence); |
+ |
+ for (size_t i = 0; i < kNumTasksPerTest; ++i) { |
+ PostTaskHelper(make_scoped_ptr(new Task( |
+ FROM_HERE, factory.CreateTaskClosure(), TaskTraits())), |
+ sequence_a, &shared_priority_queue_, &task_tracker_); |
robliao
2016/03/24 23:03:27
We definitely don't guarantee interleaving across
fdoray
2016/03/29 18:33:34
Done.
|
+ PostTaskHelper(make_scoped_ptr(new Task( |
+ FROM_HERE, factory.CreateTaskClosure(), TaskTraits())), |
+ sequence_b, &shared_priority_queue_, &task_tracker_); |
+ } |
+ |
+ worker_thread_->WakeUp(); |
+ factory.WaitUntilLastFactoredTaskRan(); |
+ WaitUntilIdle(); |
+ EXPECT_EQ(2, num_state_changes()); |
+ EXPECT_EQ(2 * (kNumTasksPerTest - 1), num_popped_task_from_shared_sequence()); |
+} |
+ |
+// Verify that a shared task and a single-threaded task run successfully when |
+// they are posted to a WorkerThread. The WorkerThread is not woken up by the |
+// test after the shared task is posted. The wake-up is done by the task runner |
+// through which the single-thread task is posted. |
+TEST_F(TaskSchedulerWorkerThreadTest, PostSharedAndSingleThreadedTasks) { |
+ TaskClosureFactory factory; |
+ |
+ // Post a task in the shared priority queue. Don't wake up the WorkerThread. |
+ PostTaskHelper(make_scoped_ptr(new Task( |
+ FROM_HERE, factory.CreateTaskClosure(), TaskTraits())), |
+ make_scoped_refptr(new Sequence), &shared_priority_queue_, |
+ &task_tracker_); |
+ |
+ // Post a task in the single-threaded priority queue. The TaskRunner will wake |
+ // up the WorkerThread. |
+ worker_thread_->CreateTaskRunnerWithTraits(TaskTraits()) |
+ ->PostTask(FROM_HERE, factory.CreateTaskClosure()); |
+ |
+ factory.WaitUntilLastFactoredTaskRan(); |
+ WaitUntilIdle(); |
+ EXPECT_EQ(2, num_state_changes()); |
+ EXPECT_EQ(0, num_popped_task_from_shared_sequence()); |
+} |
+ |
+// Verify that 500 shared tasks and 500 single-threaded tasks run successfully |
+// when they are posted to a WorkerThread. The WorkerThread is not woken up by |
+// the test after a shared task is posted. The wake-ups are done by the task |
+// runner through which the single-thread tasks are posted. |
+TEST_F(TaskSchedulerWorkerThreadTest, |
+ PostMultipleSharedAndSingleThreadedTasks) { |
+ TaskClosureFactory factory; |
+ |
+ for (size_t i = 0; i < kNumTasksPerTest; ++i) { |
+ // Post a task in the shared priority queue. Don't wake up the WorkerThread. |
+ PostTaskHelper(make_scoped_ptr(new Task( |
+ FROM_HERE, factory.CreateTaskClosure(), TaskTraits())), |
+ make_scoped_refptr(new Sequence), &shared_priority_queue_, |
+ &task_tracker_); |
+ |
+ // Post a task in the single-threaded priority queue. The TaskRunner will |
+ // wake up the WorkerThread. |
+ worker_thread_->CreateTaskRunnerWithTraits(TaskTraits()) |
+ ->PostTask(FROM_HERE, factory.CreateTaskClosure()); |
+ |
+ factory.WaitUntilLastFactoredTaskRan(); |
+ WaitUntilIdle(); |
+ EXPECT_EQ(2 * (i + 1), num_state_changes()); |
+ } |
+ |
+ EXPECT_EQ(0, num_popped_task_from_shared_sequence()); |
+} |
+ |
+// Verify that 10000 single-threaded tasks posted to a single WorkerThread from |
+// 10 threads run successfully. |
+TEST_F(TaskSchedulerWorkerThreadTest, |
+ PostSingleThreadedTasksFromMultipleThreads) { |
+ const size_t kNumThreads = 20; |
+ std::vector<scoped_ptr<ThreadPostingTasks>> threads; |
+ for (size_t i = 0; i < kNumThreads; ++i) { |
+ threads.push_back( |
+ make_scoped_ptr(new ThreadPostingTasks(worker_thread_.get()))); |
+ threads.back()->Start(); |
+ } |
+ |
+ for (const auto& thread : threads) { |
+ thread->Join(); |
+ thread->WaitUntilLastFactoredTaskRan(); |
+ } |
+ |
+ WaitUntilIdle(); |
+ EXPECT_GE(num_state_changes(), 2); |
+ EXPECT_EQ(0, num_popped_task_from_shared_sequence()); |
+} |
+ |
+} // namespace internal |
+} // namespace base |