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

Side by Side Diff: base/task_scheduler/scheduler_thread_pool_unittest.cc

Issue 1708773002: TaskScheduler [7] SchedulerThreadPool (Closed) Base URL: https://chromium.googlesource.com/chromium/src.git@s_5_worker_thread
Patch Set: self review Created 4 years, 8 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 unified diff | Download patch
OLDNEW
(Empty)
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
3 // found in the LICENSE file.
4
5 #include "base/task_scheduler/scheduler_thread_pool.h"
6
7 #include <stddef.h>
8
9 #include <memory>
10 #include <unordered_set>
11 #include <vector>
12
13 #include "base/bind.h"
14 #include "base/bind_helpers.h"
15 #include "base/macros.h"
16 #include "base/memory/ptr_util.h"
17 #include "base/memory/ref_counted.h"
18 #include "base/synchronization/condition_variable.h"
19 #include "base/synchronization/lock.h"
20 #include "base/synchronization/waitable_event.h"
21 #include "base/task_runner.h"
22 #include "base/task_scheduler/sequence.h"
23 #include "base/task_scheduler/sequence_sort_key.h"
24 #include "base/task_scheduler/task_tracker.h"
25 #include "base/threading/platform_thread.h"
26 #include "base/threading/simple_thread.h"
27 #include "testing/gtest/include/gtest/gtest.h"
28
29 namespace base {
30 namespace internal {
31 namespace {
32
33 const size_t kNumThreadsInThreadPool = 4;
34 const size_t kNumThreadsPostingTasks = 4;
35 const size_t kNumTasksPostedPerThread = 150;
36
37 class TaskSchedulerThreadPoolTest : public testing::Test {
38 protected:
39 TaskSchedulerThreadPoolTest() = default;
40
41 void SetUp() override {
42 thread_pool_ = SchedulerThreadPool::CreateThreadPool(
43 ThreadPriority::NORMAL, kNumThreadsInThreadPool,
44 Bind(&TaskSchedulerThreadPoolTest::EnqueueSequenceCallback,
45 Unretained(this)),
46 &task_tracker_);
47 ASSERT_TRUE(thread_pool_);
48 }
49
50 void TearDown() override {
51 thread_pool_->WaitForAllWorkerThreadsIdleForTesting();
52 thread_pool_->JoinForTesting();
53 }
54
55 std::unique_ptr<SchedulerThreadPool> thread_pool_;
56
57 private:
58 void EnqueueSequenceCallback(scoped_refptr<Sequence> sequence) {
59 // In production code, this callback would be implemented by the
60 // TaskScheduler which would first determine which PriorityQueue the
61 // sequence must be reinserted.
62 const SequenceSortKey sort_key(sequence->GetSortKey());
63 thread_pool_->EnqueueSequence(std::move(sequence), sort_key);
64 }
65
66 TaskTracker task_tracker_;
67
68 DISALLOW_COPY_AND_ASSIGN(TaskSchedulerThreadPoolTest);
69 };
70
71 class TaskFactory {
72 public:
73 TaskFactory() : cv_(&lock_) {}
74
75 // Posts a task through |task_runner|. If |post_nested_task| is true, the task
76 // will post a new task through |task_runner| when it runs. If |event| is set,
77 // the task will block until it is signaled.
78 void PostTestTask(scoped_refptr<TaskRunner> task_runner,
79 bool post_nested_task,
80 WaitableEvent* event) {
81 AutoLock auto_lock(lock_);
82 EXPECT_TRUE(task_runner->PostTask(
83 FROM_HERE, Bind(&TaskFactory::RunTaskCallback, Unretained(this),
84 num_created_tasks_++, task_runner, post_nested_task,
85 Unretained(event))));
86 }
87
88 // Waits for all tasks posted by PostTestTask() to start running. It is not
89 // guaranteed that the tasks have completed their execution when this returns.
90 void WaitForAllTasksToRun() const {
91 AutoLock auto_lock(lock_);
92 while (ran_tasks_.size() < num_created_tasks_)
93 cv_.Wait();
94 }
95
96 size_t NumRunTasks() const {
97 AutoLock auto_lock(lock_);
98 return ran_tasks_.size();
99 }
100
101 private:
102 void RunTaskCallback(size_t task_index,
103 scoped_refptr<TaskRunner> task_runner,
104 bool post_nested_task,
105 WaitableEvent* event) {
106 if (post_nested_task)
107 PostTestTask(task_runner, false, nullptr);
108
109 EXPECT_TRUE(task_runner->RunsTasksOnCurrentThread());
110
111 {
112 AutoLock auto_lock(lock_);
113
114 if (ran_tasks_.find(task_index) != ran_tasks_.end())
115 ADD_FAILURE() << "A task ran more than once.";
116 ran_tasks_.insert(task_index);
117
118 cv_.Signal();
119 }
120
121 if (event)
122 event->Wait();
123 }
124
125 // Synchronizes access to all members below.
126 mutable Lock lock_;
127
128 // Condition variable signaled when a task runs.
129 mutable ConditionVariable cv_;
130
131 // Number of tasks posted by PostTestTask().
132 size_t num_created_tasks_ = 0;
133
134 // Indexes of tasks that ran.
135 std::unordered_set<size_t> ran_tasks_;
136
137 DISALLOW_COPY_AND_ASSIGN(TaskFactory);
138 };
139
140 class ThreadPostingTasks : public SimpleThread {
141 public:
142 // Constructs a thread that posts tasks to |thread_pool| through an
143 // |execution_mode| task runner. If |wait_for_all_threads_idle| is true, the
144 // thread wait until all worker threads in |thread_pool| are idle before
145 // posting a new task. If |post_nested_task| is true, each task posted by this
146 // thread posts another task when it runs.
147 ThreadPostingTasks(SchedulerThreadPool* thread_pool,
148 ExecutionMode execution_mode,
149 bool wait_for_all_threads_idle,
150 bool post_nested_task)
151 : SimpleThread("ThreadPostingTasks"),
152 thread_pool_(thread_pool),
153 task_runner_(thread_pool_->CreateTaskRunnerWithTraits(TaskTraits(),
154 execution_mode)),
155 wait_for_all_threads_idle_(wait_for_all_threads_idle),
156 post_nested_task_(post_nested_task) {
157 DCHECK(thread_pool_);
158 }
159
160 const TaskFactory* factory() const { return &factory_; }
161
162 private:
163 void Run() override {
164 EXPECT_FALSE(task_runner_->RunsTasksOnCurrentThread());
165
166 for (size_t i = 0; i < kNumTasksPostedPerThread; ++i) {
167 if (wait_for_all_threads_idle_)
168 thread_pool_->WaitForAllWorkerThreadsIdleForTesting();
169 factory_.PostTestTask(task_runner_, post_nested_task_, nullptr);
170 }
171 }
172
173 SchedulerThreadPool* const thread_pool_;
174 const scoped_refptr<TaskRunner> task_runner_;
175 const bool wait_for_all_threads_idle_;
176 const bool post_nested_task_;
177 TaskFactory factory_;
178
179 DISALLOW_COPY_AND_ASSIGN(ThreadPostingTasks);
180 };
181
182 TEST_F(TaskSchedulerThreadPoolTest, PostParallelTasks) {
183 // Create threads to post tasks to PARALLEL TaskRunners.
184 std::vector<std::unique_ptr<ThreadPostingTasks>> threads_posting_tasks;
185 for (size_t i = 0; i < kNumThreadsPostingTasks; ++i) {
186 const bool kWaitForAllThreadIdle = false;
187 const bool kPostNestedTasks = false;
188 threads_posting_tasks.push_back(WrapUnique(
189 new ThreadPostingTasks(thread_pool_.get(), ExecutionMode::PARALLEL,
190 kWaitForAllThreadIdle, kPostNestedTasks)));
191 threads_posting_tasks.back()->Start();
192 }
193
194 // Wait for all tasks to run.
195 for (const auto& thread_posting_tasks : threads_posting_tasks) {
196 thread_posting_tasks->Join();
197 thread_posting_tasks->factory()->WaitForAllTasksToRun();
198 EXPECT_EQ(kNumTasksPostedPerThread,
199 thread_posting_tasks->factory()->NumRunTasks());
200 }
201
202 // Wait until all worker threads are idle to be sure that no task accesses
203 // its TaskFactory after |thread_posting_tasks| is destroyed.
204 thread_pool_->WaitForAllWorkerThreadsIdleForTesting();
205 }
206
207 TEST_F(TaskSchedulerThreadPoolTest, PostParallelTasksWaitAllThreadsIdle) {
208 // Create threads to post tasks to PARALLEL TaskRunners. To verify that
209 // worker threads can sleep and be woken up when new tasks are posted, wait
210 // for all threads to become idle before posting a new task.
211 std::vector<std::unique_ptr<ThreadPostingTasks>> threads_posting_tasks;
212 for (size_t i = 0; i < kNumThreadsPostingTasks; ++i) {
213 const bool kWaitForAllThreadIdle = true;
214 const bool kPostNestedTasks = false;
215 threads_posting_tasks.push_back(WrapUnique(
216 new ThreadPostingTasks(thread_pool_.get(), ExecutionMode::PARALLEL,
217 kWaitForAllThreadIdle, kPostNestedTasks)));
218 threads_posting_tasks.back()->Start();
219 }
220
221 // Wait for all tasks to run.
222 for (const auto& thread_posting_tasks : threads_posting_tasks) {
223 thread_posting_tasks->Join();
224 thread_posting_tasks->factory()->WaitForAllTasksToRun();
225 EXPECT_EQ(kNumTasksPostedPerThread,
226 thread_posting_tasks->factory()->NumRunTasks());
227 }
228
229 // Wait until all worker threads are idle to be sure that no task accesses
230 // its TaskFactory after |thread_posting_tasks| is destroyed.
231 thread_pool_->WaitForAllWorkerThreadsIdleForTesting();
232 }
233
234 TEST_F(TaskSchedulerThreadPoolTest, NestedPostParallelTasks) {
235 // Create threads to post tasks to PARALLEL TaskRunners. Each task posted by
236 // these threads will post another task when it runs.
237 std::vector<std::unique_ptr<ThreadPostingTasks>> threads_posting_tasks;
238 for (size_t i = 0; i < kNumThreadsPostingTasks; ++i) {
239 const bool kWaitForAllThreadIdle = false;
240 const bool kPostNestedTasks = true;
241 threads_posting_tasks.push_back(WrapUnique(
242 new ThreadPostingTasks(thread_pool_.get(), ExecutionMode::PARALLEL,
243 kWaitForAllThreadIdle, kPostNestedTasks)));
244 threads_posting_tasks.back()->Start();
245 }
246
247 // Wait for all tasks to run.
248 for (const auto& thread_posting_tasks : threads_posting_tasks) {
249 thread_posting_tasks->Join();
250 thread_posting_tasks->factory()->WaitForAllTasksToRun();
251 EXPECT_EQ(2 * kNumTasksPostedPerThread,
252 thread_posting_tasks->factory()->NumRunTasks());
253 }
254
255 // Wait until all worker threads are idle to be sure that no task accesses
256 // its TaskFactory after |thread_posting_tasks| is destroyed.
257 thread_pool_->WaitForAllWorkerThreadsIdleForTesting();
258 }
259
260 TEST_F(TaskSchedulerThreadPoolTest, PostParallelTasksWithOneAvailableThread) {
261 TaskFactory factory;
262
263 // Post tasks to keep all threads busy except one until |event| is signaled.
264 WaitableEvent event(true, false);
265 auto task_runner = thread_pool_->CreateTaskRunnerWithTraits(
266 TaskTraits(), ExecutionMode::PARALLEL);
267 for (size_t i = 0; i < (kNumThreadsInThreadPool - 1); ++i)
268 factory.PostTestTask(task_runner, false, &event);
269 factory.WaitForAllTasksToRun();
270
271 // Post |kNumTasksPostedPerThread| tasks that should all run despite the fact
272 // that only one thread in |thread_pool_| isn't busy.
273 for (size_t i = 0; i < kNumTasksPostedPerThread; ++i)
274 factory.PostTestTask(task_runner, false, nullptr);
275 factory.WaitForAllTasksToRun();
276
277 // Release tasks waiting on |event|.
278 event.Signal();
279
280 // Wait until all worker threads are idle to be sure that no task accesses
281 // |factory| after it is destroyed.
282 thread_pool_->WaitForAllWorkerThreadsIdleForTesting();
283 }
284
285 TEST_F(TaskSchedulerThreadPoolTest, Saturate) {
286 TaskFactory factory;
287
288 // Verify that it is possible to have |kNumThreadsInThreadPool| tasks running
289 // simultaneously.
290 WaitableEvent event(true, false);
291 auto task_runner = thread_pool_->CreateTaskRunnerWithTraits(
292 TaskTraits(), ExecutionMode::PARALLEL);
293 for (size_t i = 0; i < kNumThreadsInThreadPool; ++i)
294 factory.PostTestTask(task_runner, false, &event);
295 factory.WaitForAllTasksToRun();
296
297 // Release tasks waiting on |event|.
298 event.Signal();
299
300 // Wait until all worker threads are idle to be sure that no task accesses
301 // |factory| after it is destroyed.
302 thread_pool_->WaitForAllWorkerThreadsIdleForTesting();
303 }
304
305 } // namespace
306 } // namespace internal
307 } // namespace base
OLDNEW

Powered by Google App Engine
This is Rietveld 408576698