1 // Copyright (c) 2012 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.
5 #include "base/threading/sequenced_worker_pool.h"
10 #include "base/compiler_specific.h"
11 #include "base/memory/ref_counted.h"
12 #include "base/memory/scoped_ptr.h"
13 #include "base/message_loop/message_loop.h"
14 #include "base/message_loop/message_loop_proxy.h"
15 #include "base/synchronization/condition_variable.h"
16 #include "base/synchronization/lock.h"
17 #include "base/test/sequenced_task_runner_test_template.h"
18 #include "base/test/sequenced_worker_pool_owner.h"
19 #include "base/test/task_runner_test_template.h"
20 #include "base/test/test_timeouts.h"
21 #include "base/threading/platform_thread.h"
22 #include "base/time/time.h"
23 #include "base/tracked_objects.h"
24 #include "testing/gtest/include/gtest/gtest.h"
30 // Many of these tests have failure modes where they'll hang forever. These
31 // tests should not be flaky, and hanging indicates a type of failure. Do not
32 // mark as flaky if they're hanging, it's likely an actual bug.
36 const size_t kNumWorkerThreads
= 3;
38 // Allows a number of threads to all be blocked on the same event, and
39 // provides a way to unblock a certain number of them.
42 ThreadBlocker() : lock_(), cond_var_(&lock_
), unblock_counter_(0) {}
46 base::AutoLock
lock(lock_
);
47 while (unblock_counter_
== 0)
54 void Unblock(size_t count
) {
56 base::AutoLock
lock(lock_
);
57 DCHECK(unblock_counter_
== 0);
58 unblock_counter_
= count
;
65 base::ConditionVariable cond_var_
;
67 size_t unblock_counter_
;
70 class DestructionDeadlockChecker
71 : public base::RefCountedThreadSafe
<DestructionDeadlockChecker
> {
73 DestructionDeadlockChecker(const scoped_refptr
<SequencedWorkerPool
>& pool
)
77 virtual ~DestructionDeadlockChecker() {
78 // This method should not deadlock.
79 pool_
->RunsTasksOnCurrentThread();
83 scoped_refptr
<SequencedWorkerPool
> pool_
;
84 friend class base::RefCountedThreadSafe
<DestructionDeadlockChecker
>;
87 class TestTracker
: public base::RefCountedThreadSafe
<TestTracker
> {
95 // Each of these tasks appends the argument to the complete sequence vector
96 // so calling code can see what order they finished in.
97 void FastTask(int id
) {
101 void SlowTask(int id
) {
102 base::PlatformThread::Sleep(base::TimeDelta::FromSeconds(1));
103 SignalWorkerDone(id
);
106 void BlockTask(int id
, ThreadBlocker
* blocker
) {
107 // Note that this task has started and signal anybody waiting for that
110 base::AutoLock
lock(lock_
);
116 SignalWorkerDone(id
);
119 void PostAdditionalTasks(
120 int id
, SequencedWorkerPool
* pool
,
121 bool expected_return_value
) {
122 Closure fast_task
= base::Bind(&TestTracker::FastTask
, this, 100);
123 EXPECT_EQ(expected_return_value
,
124 pool
->PostWorkerTaskWithShutdownBehavior(
125 FROM_HERE
, fast_task
,
126 SequencedWorkerPool::CONTINUE_ON_SHUTDOWN
));
127 EXPECT_EQ(expected_return_value
,
128 pool
->PostWorkerTaskWithShutdownBehavior(
129 FROM_HERE
, fast_task
,
130 SequencedWorkerPool::SKIP_ON_SHUTDOWN
));
131 pool
->PostWorkerTaskWithShutdownBehavior(
132 FROM_HERE
, fast_task
,
133 SequencedWorkerPool::BLOCK_SHUTDOWN
);
134 SignalWorkerDone(id
);
137 // This task posts itself back onto the SequencedWorkerPool before it
138 // finishes running. Each instance of the task maintains a strong reference
139 // to a DestructionDeadlockChecker. The DestructionDeadlockChecker is only
140 // destroyed when the task is destroyed without being run, which only happens
141 // during destruction of the SequencedWorkerPool.
142 void PostRepostingTask(
143 const scoped_refptr
<SequencedWorkerPool
>& pool
,
144 const scoped_refptr
<DestructionDeadlockChecker
>& checker
) {
145 Closure reposting_task
=
146 base::Bind(&TestTracker::PostRepostingTask
, this, pool
, checker
);
147 pool
->PostWorkerTaskWithShutdownBehavior(
148 FROM_HERE
, reposting_task
, SequencedWorkerPool::SKIP_ON_SHUTDOWN
);
151 // Waits until the given number of tasks have started executing.
152 void WaitUntilTasksBlocked(size_t count
) {
154 base::AutoLock
lock(lock_
);
155 while (started_events_
< count
)
161 // Blocks the current thread until at least the given number of tasks are in
162 // the completed vector, and then returns a copy.
163 std::vector
<int> WaitUntilTasksComplete(size_t num_tasks
) {
164 std::vector
<int> ret
;
166 base::AutoLock
lock(lock_
);
167 while (complete_sequence_
.size() < num_tasks
)
169 ret
= complete_sequence_
;
175 size_t GetTasksCompletedCount() {
176 base::AutoLock
lock(lock_
);
177 return complete_sequence_
.size();
180 void ClearCompleteSequence() {
181 base::AutoLock
lock(lock_
);
182 complete_sequence_
.clear();
187 friend class base::RefCountedThreadSafe
<TestTracker
>;
190 void SignalWorkerDone(int id
) {
192 base::AutoLock
lock(lock_
);
193 complete_sequence_
.push_back(id
);
198 // Protects the complete_sequence.
201 base::ConditionVariable cond_var_
;
203 // Protected by lock_.
204 std::vector
<int> complete_sequence_
;
206 // Counter of the number of "block" workers that have started.
207 size_t started_events_
;
210 class SequencedWorkerPoolTest
: public testing::Test
{
212 SequencedWorkerPoolTest()
213 : tracker_(new TestTracker
) {
217 void TearDown() override
{ pool()->Shutdown(); }
219 const scoped_refptr
<SequencedWorkerPool
>& pool() {
220 return pool_owner_
->pool();
222 TestTracker
* tracker() { return tracker_
.get(); }
224 // Destroys the SequencedWorkerPool instance, blocking until it is fully shut
225 // down, and creates a new instance.
227 pool_owner_
.reset(new SequencedWorkerPoolOwner(kNumWorkerThreads
, "test"));
230 void SetWillWaitForShutdownCallback(const Closure
& callback
) {
231 pool_owner_
->SetWillWaitForShutdownCallback(callback
);
234 // Ensures that the given number of worker threads is created by adding
235 // tasks and waiting until they complete. Worker thread creation is
236 // serialized, can happen on background threads asynchronously, and doesn't
237 // happen any more at shutdown. This means that if a test posts a bunch of
238 // tasks and calls shutdown, fewer workers will be created than the test may
241 // This function ensures that this condition can't happen so tests can make
242 // assumptions about the number of workers active. See the comment in
243 // PrepareToStartAdditionalThreadIfNecessary in the .cc file for more
246 // It will post tasks to the queue with id -1. It also assumes this is the
247 // first thing called in a test since it will clear the complete_sequence_.
248 void EnsureAllWorkersCreated() {
249 // Create a bunch of threads, all waiting. This will cause that may
250 // workers to be created.
251 ThreadBlocker blocker
;
252 for (size_t i
= 0; i
< kNumWorkerThreads
; i
++) {
253 pool()->PostWorkerTask(FROM_HERE
,
254 base::Bind(&TestTracker::BlockTask
,
255 tracker(), -1, &blocker
));
257 tracker()->WaitUntilTasksBlocked(kNumWorkerThreads
);
259 // Now wake them up and wait until they're done.
260 blocker
.Unblock(kNumWorkerThreads
);
261 tracker()->WaitUntilTasksComplete(kNumWorkerThreads
);
263 // Clean up the task IDs we added.
264 tracker()->ClearCompleteSequence();
267 int has_work_call_count() const {
268 return pool_owner_
->has_work_call_count();
272 MessageLoop message_loop_
;
273 scoped_ptr
<SequencedWorkerPoolOwner
> pool_owner_
;
274 const scoped_refptr
<TestTracker
> tracker_
;
277 // Checks that the given number of entries are in the tasks to complete of
278 // the given tracker, and then signals the given event the given number of
279 // times. This is used to wakt up blocked background threads before blocking
281 void EnsureTasksToCompleteCountAndUnblock(scoped_refptr
<TestTracker
> tracker
,
282 size_t expected_tasks_to_complete
,
283 ThreadBlocker
* blocker
,
284 size_t threads_to_awake
) {
286 expected_tasks_to_complete
,
287 tracker
->WaitUntilTasksComplete(expected_tasks_to_complete
).size());
289 blocker
->Unblock(threads_to_awake
);
292 class DeletionHelper
: public base::RefCountedThreadSafe
<DeletionHelper
> {
294 explicit DeletionHelper(
295 const scoped_refptr
<base::RefCountedData
<bool> >& deleted_flag
)
296 : deleted_flag_(deleted_flag
) {
300 friend class base::RefCountedThreadSafe
<DeletionHelper
>;
301 virtual ~DeletionHelper() { deleted_flag_
->data
= true; }
303 const scoped_refptr
<base::RefCountedData
<bool> > deleted_flag_
;
304 DISALLOW_COPY_AND_ASSIGN(DeletionHelper
);
307 void HoldPoolReference(const scoped_refptr
<base::SequencedWorkerPool
>& pool
,
308 const scoped_refptr
<DeletionHelper
>& helper
) {
309 ADD_FAILURE() << "Should never run";
312 // Tests that delayed tasks are deleted upon shutdown of the pool.
313 TEST_F(SequencedWorkerPoolTest
, DelayedTaskDuringShutdown
) {
314 // Post something to verify the pool is started up.
315 EXPECT_TRUE(pool()->PostTask(
316 FROM_HERE
, base::Bind(&TestTracker::FastTask
, tracker(), 1)));
318 scoped_refptr
<base::RefCountedData
<bool> > deleted_flag(
319 new base::RefCountedData
<bool>(false));
321 base::Time
posted_at(base::Time::Now());
322 // Post something that shouldn't run.
323 EXPECT_TRUE(pool()->PostDelayedTask(
325 base::Bind(&HoldPoolReference
,
327 make_scoped_refptr(new DeletionHelper(deleted_flag
))),
328 TestTimeouts::action_timeout()));
330 std::vector
<int> completion_sequence
= tracker()->WaitUntilTasksComplete(1);
331 ASSERT_EQ(1u, completion_sequence
.size());
332 ASSERT_EQ(1, completion_sequence
[0]);
335 // Shutdown is asynchronous, so use ResetPool() to block until the pool is
336 // fully destroyed (and thus shut down).
339 // Verify that we didn't block until the task was due.
340 ASSERT_LT(base::Time::Now() - posted_at
, TestTimeouts::action_timeout());
342 // Verify that the deferred task has not only not run, but has also been
344 ASSERT_TRUE(deleted_flag
->data
);
347 // Tests that same-named tokens have the same ID.
348 TEST_F(SequencedWorkerPoolTest
, NamedTokens
) {
349 const std::string
name1("hello");
350 SequencedWorkerPool::SequenceToken token1
=
351 pool()->GetNamedSequenceToken(name1
);
353 SequencedWorkerPool::SequenceToken token2
= pool()->GetSequenceToken();
355 const std::string
name3("goodbye");
356 SequencedWorkerPool::SequenceToken token3
=
357 pool()->GetNamedSequenceToken(name3
);
359 // All 3 tokens should be different.
360 EXPECT_FALSE(token1
.Equals(token2
));
361 EXPECT_FALSE(token1
.Equals(token3
));
362 EXPECT_FALSE(token2
.Equals(token3
));
364 // Requesting the same name again should give the same value.
365 SequencedWorkerPool::SequenceToken token1again
=
366 pool()->GetNamedSequenceToken(name1
);
367 EXPECT_TRUE(token1
.Equals(token1again
));
369 SequencedWorkerPool::SequenceToken token3again
=
370 pool()->GetNamedSequenceToken(name3
);
371 EXPECT_TRUE(token3
.Equals(token3again
));
374 // Tests that posting a bunch of tasks (many more than the number of worker
375 // threads) runs them all.
376 TEST_F(SequencedWorkerPoolTest
, LotsOfTasks
) {
377 pool()->PostWorkerTask(FROM_HERE
,
378 base::Bind(&TestTracker::SlowTask
, tracker(), 0));
380 const size_t kNumTasks
= 20;
381 for (size_t i
= 1; i
< kNumTasks
; i
++) {
382 pool()->PostWorkerTask(FROM_HERE
,
383 base::Bind(&TestTracker::FastTask
, tracker(), i
));
386 std::vector
<int> result
= tracker()->WaitUntilTasksComplete(kNumTasks
);
387 EXPECT_EQ(kNumTasks
, result
.size());
390 // Tests that posting a bunch of tasks (many more than the number of
391 // worker threads) to two pools simultaneously runs them all twice.
392 // This test is meant to shake out any concurrency issues between
393 // pools (like histograms).
394 TEST_F(SequencedWorkerPoolTest
, LotsOfTasksTwoPools
) {
395 SequencedWorkerPoolOwner
pool1(kNumWorkerThreads
, "test1");
396 SequencedWorkerPoolOwner
pool2(kNumWorkerThreads
, "test2");
398 base::Closure slow_task
= base::Bind(&TestTracker::SlowTask
, tracker(), 0);
399 pool1
.pool()->PostWorkerTask(FROM_HERE
, slow_task
);
400 pool2
.pool()->PostWorkerTask(FROM_HERE
, slow_task
);
402 const size_t kNumTasks
= 20;
403 for (size_t i
= 1; i
< kNumTasks
; i
++) {
404 base::Closure fast_task
=
405 base::Bind(&TestTracker::FastTask
, tracker(), i
);
406 pool1
.pool()->PostWorkerTask(FROM_HERE
, fast_task
);
407 pool2
.pool()->PostWorkerTask(FROM_HERE
, fast_task
);
410 std::vector
<int> result
=
411 tracker()->WaitUntilTasksComplete(2*kNumTasks
);
412 EXPECT_EQ(2 * kNumTasks
, result
.size());
414 pool2
.pool()->Shutdown();
415 pool1
.pool()->Shutdown();
418 // Test that tasks with the same sequence token are executed in order but don't
419 // affect other tasks.
420 TEST_F(SequencedWorkerPoolTest
, Sequence
) {
421 // Fill all the worker threads except one.
422 const size_t kNumBackgroundTasks
= kNumWorkerThreads
- 1;
423 ThreadBlocker background_blocker
;
424 for (size_t i
= 0; i
< kNumBackgroundTasks
; i
++) {
425 pool()->PostWorkerTask(FROM_HERE
,
426 base::Bind(&TestTracker::BlockTask
,
427 tracker(), i
, &background_blocker
));
429 tracker()->WaitUntilTasksBlocked(kNumBackgroundTasks
);
431 // Create two tasks with the same sequence token, one that will block on the
432 // event, and one which will just complete quickly when it's run. Since there
433 // is one worker thread free, the first task will start and then block, and
434 // the second task should be waiting.
435 ThreadBlocker blocker
;
436 SequencedWorkerPool::SequenceToken token1
= pool()->GetSequenceToken();
437 pool()->PostSequencedWorkerTask(
439 base::Bind(&TestTracker::BlockTask
, tracker(), 100, &blocker
));
440 pool()->PostSequencedWorkerTask(
442 base::Bind(&TestTracker::FastTask
, tracker(), 101));
443 EXPECT_EQ(0u, tracker()->WaitUntilTasksComplete(0).size());
445 // Create another two tasks as above with a different token. These will be
446 // blocked since there are no slots to run.
447 SequencedWorkerPool::SequenceToken token2
= pool()->GetSequenceToken();
448 pool()->PostSequencedWorkerTask(
450 base::Bind(&TestTracker::FastTask
, tracker(), 200));
451 pool()->PostSequencedWorkerTask(
453 base::Bind(&TestTracker::FastTask
, tracker(), 201));
454 EXPECT_EQ(0u, tracker()->WaitUntilTasksComplete(0).size());
456 // Let one background task complete. This should then let both tasks of
457 // token2 run to completion in order. The second task of token1 should still
459 background_blocker
.Unblock(1);
460 std::vector
<int> result
= tracker()->WaitUntilTasksComplete(3);
461 ASSERT_EQ(3u, result
.size());
462 EXPECT_EQ(200, result
[1]);
463 EXPECT_EQ(201, result
[2]);
465 // Finish the rest of the background tasks. This should leave some workers
466 // free with the second token1 task still blocked on the first.
467 background_blocker
.Unblock(kNumBackgroundTasks
- 1);
468 EXPECT_EQ(kNumBackgroundTasks
+ 2,
469 tracker()->WaitUntilTasksComplete(kNumBackgroundTasks
+ 2).size());
471 // Allow the first task of token1 to complete. This should run the second.
473 result
= tracker()->WaitUntilTasksComplete(kNumBackgroundTasks
+ 4);
474 ASSERT_EQ(kNumBackgroundTasks
+ 4, result
.size());
475 EXPECT_EQ(100, result
[result
.size() - 2]);
476 EXPECT_EQ(101, result
[result
.size() - 1]);
479 // Tests that any tasks posted after Shutdown are ignored.
480 // Disabled for flakiness. See http://crbug.com/166451.
481 TEST_F(SequencedWorkerPoolTest
, DISABLED_IgnoresAfterShutdown
) {
482 // Start tasks to take all the threads and block them.
483 EnsureAllWorkersCreated();
484 ThreadBlocker blocker
;
485 for (size_t i
= 0; i
< kNumWorkerThreads
; i
++) {
486 pool()->PostWorkerTask(FROM_HERE
,
487 base::Bind(&TestTracker::BlockTask
,
488 tracker(), i
, &blocker
));
490 tracker()->WaitUntilTasksBlocked(kNumWorkerThreads
);
492 SetWillWaitForShutdownCallback(
493 base::Bind(&EnsureTasksToCompleteCountAndUnblock
,
494 scoped_refptr
<TestTracker
>(tracker()), 0,
495 &blocker
, kNumWorkerThreads
));
497 // Shutdown the worker pool. This should discard all non-blocking tasks.
498 const int kMaxNewBlockingTasksAfterShutdown
= 100;
499 pool()->Shutdown(kMaxNewBlockingTasksAfterShutdown
);
501 int old_has_work_call_count
= has_work_call_count();
503 std::vector
<int> result
=
504 tracker()->WaitUntilTasksComplete(kNumWorkerThreads
);
506 // The kNumWorkerThread items should have completed, in no particular order.
507 ASSERT_EQ(kNumWorkerThreads
, result
.size());
508 for (size_t i
= 0; i
< kNumWorkerThreads
; i
++) {
509 EXPECT_TRUE(std::find(result
.begin(), result
.end(), static_cast<int>(i
)) !=
513 // No further tasks, regardless of shutdown mode, should be allowed.
514 EXPECT_FALSE(pool()->PostWorkerTaskWithShutdownBehavior(
516 base::Bind(&TestTracker::FastTask
, tracker(), 100),
517 SequencedWorkerPool::CONTINUE_ON_SHUTDOWN
));
518 EXPECT_FALSE(pool()->PostWorkerTaskWithShutdownBehavior(
520 base::Bind(&TestTracker::FastTask
, tracker(), 101),
521 SequencedWorkerPool::SKIP_ON_SHUTDOWN
));
522 EXPECT_FALSE(pool()->PostWorkerTaskWithShutdownBehavior(
524 base::Bind(&TestTracker::FastTask
, tracker(), 102),
525 SequencedWorkerPool::BLOCK_SHUTDOWN
));
527 ASSERT_EQ(old_has_work_call_count
, has_work_call_count());
530 TEST_F(SequencedWorkerPoolTest
, AllowsAfterShutdown
) {
531 // Test that <n> new blocking tasks are allowed provided they're posted
532 // by a running tasks.
533 EnsureAllWorkersCreated();
534 ThreadBlocker blocker
;
536 // Start tasks to take all the threads and block them.
537 const int kNumBlockTasks
= static_cast<int>(kNumWorkerThreads
);
538 for (int i
= 0; i
< kNumBlockTasks
; ++i
) {
539 EXPECT_TRUE(pool()->PostWorkerTask(
541 base::Bind(&TestTracker::BlockTask
, tracker(), i
, &blocker
)));
543 tracker()->WaitUntilTasksBlocked(kNumWorkerThreads
);
545 // Queue up shutdown blocking tasks behind those which will attempt to post
546 // additional tasks when run, PostAdditionalTasks attemtps to post 3
547 // new FastTasks, one for each shutdown_behavior.
548 const int kNumQueuedTasks
= static_cast<int>(kNumWorkerThreads
);
549 for (int i
= 0; i
< kNumQueuedTasks
; ++i
) {
550 EXPECT_TRUE(pool()->PostWorkerTaskWithShutdownBehavior(
552 base::Bind(&TestTracker::PostAdditionalTasks
, tracker(), i
, pool(),
554 SequencedWorkerPool::BLOCK_SHUTDOWN
));
557 // Setup to open the floodgates from within Shutdown().
558 SetWillWaitForShutdownCallback(
559 base::Bind(&EnsureTasksToCompleteCountAndUnblock
,
560 scoped_refptr
<TestTracker
>(tracker()),
561 0, &blocker
, kNumBlockTasks
));
563 // Allow half of the additional blocking tasks thru.
564 const int kNumNewBlockingTasksToAllow
= kNumWorkerThreads
/ 2;
565 pool()->Shutdown(kNumNewBlockingTasksToAllow
);
567 // Ensure that the correct number of tasks actually got run.
568 tracker()->WaitUntilTasksComplete(static_cast<size_t>(
569 kNumBlockTasks
+ kNumQueuedTasks
+ kNumNewBlockingTasksToAllow
));
571 // Clean up the task IDs we added and go home.
572 tracker()->ClearCompleteSequence();
575 // Tests that unrun tasks are discarded properly according to their shutdown
577 TEST_F(SequencedWorkerPoolTest
, DiscardOnShutdown
) {
578 // Start tasks to take all the threads and block them.
579 EnsureAllWorkersCreated();
580 ThreadBlocker blocker
;
581 for (size_t i
= 0; i
< kNumWorkerThreads
; i
++) {
582 pool()->PostWorkerTask(FROM_HERE
,
583 base::Bind(&TestTracker::BlockTask
,
584 tracker(), i
, &blocker
));
586 tracker()->WaitUntilTasksBlocked(kNumWorkerThreads
);
588 // Create some tasks with different shutdown modes.
589 pool()->PostWorkerTaskWithShutdownBehavior(
591 base::Bind(&TestTracker::FastTask
, tracker(), 100),
592 SequencedWorkerPool::CONTINUE_ON_SHUTDOWN
);
593 pool()->PostWorkerTaskWithShutdownBehavior(
595 base::Bind(&TestTracker::FastTask
, tracker(), 101),
596 SequencedWorkerPool::SKIP_ON_SHUTDOWN
);
597 pool()->PostWorkerTaskWithShutdownBehavior(
599 base::Bind(&TestTracker::FastTask
, tracker(), 102),
600 SequencedWorkerPool::BLOCK_SHUTDOWN
);
602 // Shutdown the worker pool. This should discard all non-blocking tasks.
603 SetWillWaitForShutdownCallback(
604 base::Bind(&EnsureTasksToCompleteCountAndUnblock
,
605 scoped_refptr
<TestTracker
>(tracker()), 0,
606 &blocker
, kNumWorkerThreads
));
609 std::vector
<int> result
=
610 tracker()->WaitUntilTasksComplete(kNumWorkerThreads
+ 1);
612 // The kNumWorkerThread items should have completed, plus the BLOCK_SHUTDOWN
613 // one, in no particular order.
614 ASSERT_EQ(kNumWorkerThreads
+ 1, result
.size());
615 for (size_t i
= 0; i
< kNumWorkerThreads
; i
++) {
616 EXPECT_TRUE(std::find(result
.begin(), result
.end(), static_cast<int>(i
)) !=
619 EXPECT_TRUE(std::find(result
.begin(), result
.end(), 102) != result
.end());
622 // Tests that CONTINUE_ON_SHUTDOWN tasks don't block shutdown.
623 TEST_F(SequencedWorkerPoolTest
, ContinueOnShutdown
) {
624 scoped_refptr
<TaskRunner
> runner(pool()->GetTaskRunnerWithShutdownBehavior(
625 SequencedWorkerPool::CONTINUE_ON_SHUTDOWN
));
626 scoped_refptr
<SequencedTaskRunner
> sequenced_runner(
627 pool()->GetSequencedTaskRunnerWithShutdownBehavior(
628 pool()->GetSequenceToken(),
629 SequencedWorkerPool::CONTINUE_ON_SHUTDOWN
));
630 EnsureAllWorkersCreated();
631 ThreadBlocker blocker
;
632 pool()->PostWorkerTaskWithShutdownBehavior(
634 base::Bind(&TestTracker::BlockTask
,
635 tracker(), 0, &blocker
),
636 SequencedWorkerPool::CONTINUE_ON_SHUTDOWN
);
639 base::Bind(&TestTracker::BlockTask
,
640 tracker(), 1, &blocker
));
641 sequenced_runner
->PostTask(
643 base::Bind(&TestTracker::BlockTask
,
644 tracker(), 2, &blocker
));
646 tracker()->WaitUntilTasksBlocked(3);
648 // This should not block. If this test hangs, it means it failed.
651 // The task should not have completed yet.
652 EXPECT_EQ(0u, tracker()->WaitUntilTasksComplete(0).size());
654 // Posting more tasks should fail.
655 EXPECT_FALSE(pool()->PostWorkerTaskWithShutdownBehavior(
656 FROM_HERE
, base::Bind(&TestTracker::FastTask
, tracker(), 0),
657 SequencedWorkerPool::CONTINUE_ON_SHUTDOWN
));
658 EXPECT_FALSE(runner
->PostTask(
659 FROM_HERE
, base::Bind(&TestTracker::FastTask
, tracker(), 0)));
660 EXPECT_FALSE(sequenced_runner
->PostTask(
661 FROM_HERE
, base::Bind(&TestTracker::FastTask
, tracker(), 0)));
663 // Continue the background thread and make sure the tasks can complete.
665 std::vector
<int> result
= tracker()->WaitUntilTasksComplete(3);
666 EXPECT_EQ(3u, result
.size());
669 // Tests that SKIP_ON_SHUTDOWN tasks that have been started block Shutdown
670 // until they stop, but tasks not yet started do not.
671 TEST_F(SequencedWorkerPoolTest
, SkipOnShutdown
) {
672 // Start tasks to take all the threads and block them.
673 EnsureAllWorkersCreated();
674 ThreadBlocker blocker
;
676 // Now block all the threads with SKIP_ON_SHUTDOWN. Shutdown() should not
677 // return until these tasks have completed.
678 for (size_t i
= 0; i
< kNumWorkerThreads
; i
++) {
679 pool()->PostWorkerTaskWithShutdownBehavior(
681 base::Bind(&TestTracker::BlockTask
, tracker(), i
, &blocker
),
682 SequencedWorkerPool::SKIP_ON_SHUTDOWN
);
684 tracker()->WaitUntilTasksBlocked(kNumWorkerThreads
);
686 // Now post an additional task as SKIP_ON_SHUTDOWN, which should not be
687 // executed once Shutdown() has been called.
688 pool()->PostWorkerTaskWithShutdownBehavior(
690 base::Bind(&TestTracker::BlockTask
,
691 tracker(), 0, &blocker
),
692 SequencedWorkerPool::SKIP_ON_SHUTDOWN
);
694 // This callback will only be invoked if SKIP_ON_SHUTDOWN tasks that have
695 // been started block shutdown.
696 SetWillWaitForShutdownCallback(
697 base::Bind(&EnsureTasksToCompleteCountAndUnblock
,
698 scoped_refptr
<TestTracker
>(tracker()), 0,
699 &blocker
, kNumWorkerThreads
));
701 // No tasks should have completed yet.
702 EXPECT_EQ(0u, tracker()->WaitUntilTasksComplete(0).size());
704 // This should not block. If this test hangs, it means it failed.
707 // Shutdown should not return until all of the tasks have completed.
708 std::vector
<int> result
=
709 tracker()->WaitUntilTasksComplete(kNumWorkerThreads
);
711 // Only tasks marked SKIP_ON_SHUTDOWN that were already started should be
712 // allowed to complete. No additional non-blocking tasks should have been
714 ASSERT_EQ(kNumWorkerThreads
, result
.size());
715 for (size_t i
= 0; i
< kNumWorkerThreads
; i
++) {
716 EXPECT_TRUE(std::find(result
.begin(), result
.end(), static_cast<int>(i
)) !=
721 // Ensure all worker threads are created, and then trigger a spurious
722 // work signal. This shouldn't cause any other work signals to be
723 // triggered. This is a regression test for http://crbug.com/117469.
724 TEST_F(SequencedWorkerPoolTest
, SpuriousWorkSignal
) {
725 EnsureAllWorkersCreated();
726 int old_has_work_call_count
= has_work_call_count();
727 pool()->SignalHasWorkForTesting();
728 // This is inherently racy, but can only produce false positives.
729 base::PlatformThread::Sleep(base::TimeDelta::FromMilliseconds(100));
730 EXPECT_EQ(old_has_work_call_count
+ 1, has_work_call_count());
733 void IsRunningOnCurrentThreadTask(
734 SequencedWorkerPool::SequenceToken test_positive_token
,
735 SequencedWorkerPool::SequenceToken test_negative_token
,
736 SequencedWorkerPool
* pool
,
737 SequencedWorkerPool
* unused_pool
) {
738 EXPECT_TRUE(pool
->RunsTasksOnCurrentThread());
739 EXPECT_TRUE(pool
->IsRunningSequenceOnCurrentThread(test_positive_token
));
740 EXPECT_FALSE(pool
->IsRunningSequenceOnCurrentThread(test_negative_token
));
741 EXPECT_FALSE(unused_pool
->RunsTasksOnCurrentThread());
743 unused_pool
->IsRunningSequenceOnCurrentThread(test_positive_token
));
745 unused_pool
->IsRunningSequenceOnCurrentThread(test_negative_token
));
748 // Verify correctness of the IsRunningSequenceOnCurrentThread method.
749 TEST_F(SequencedWorkerPoolTest
, IsRunningOnCurrentThread
) {
750 SequencedWorkerPool::SequenceToken token1
= pool()->GetSequenceToken();
751 SequencedWorkerPool::SequenceToken token2
= pool()->GetSequenceToken();
752 SequencedWorkerPool::SequenceToken unsequenced_token
;
754 scoped_refptr
<SequencedWorkerPool
> unused_pool
=
755 new SequencedWorkerPool(2, "unused_pool");
757 EXPECT_FALSE(pool()->RunsTasksOnCurrentThread());
758 EXPECT_FALSE(pool()->IsRunningSequenceOnCurrentThread(token1
));
759 EXPECT_FALSE(pool()->IsRunningSequenceOnCurrentThread(token2
));
760 EXPECT_FALSE(pool()->IsRunningSequenceOnCurrentThread(unsequenced_token
));
761 EXPECT_FALSE(unused_pool
->RunsTasksOnCurrentThread());
762 EXPECT_FALSE(unused_pool
->IsRunningSequenceOnCurrentThread(token1
));
763 EXPECT_FALSE(unused_pool
->IsRunningSequenceOnCurrentThread(token2
));
765 unused_pool
->IsRunningSequenceOnCurrentThread(unsequenced_token
));
767 pool()->PostSequencedWorkerTask(
769 base::Bind(&IsRunningOnCurrentThreadTask
,
770 token1
, token2
, pool(), unused_pool
));
771 pool()->PostSequencedWorkerTask(
773 base::Bind(&IsRunningOnCurrentThreadTask
,
774 token2
, unsequenced_token
, pool(), unused_pool
));
775 pool()->PostWorkerTask(
777 base::Bind(&IsRunningOnCurrentThreadTask
,
778 unsequenced_token
, token1
, pool(), unused_pool
));
780 unused_pool
->Shutdown();
783 // Checks that tasks are destroyed in the right context during shutdown. If a
784 // task is destroyed while SequencedWorkerPool's global lock is held,
785 // SequencedWorkerPool might deadlock.
786 TEST_F(SequencedWorkerPoolTest
, AvoidsDeadlockOnShutdown
) {
787 for (int i
= 0; i
< 4; ++i
) {
788 scoped_refptr
<DestructionDeadlockChecker
> checker(
789 new DestructionDeadlockChecker(pool()));
790 tracker()->PostRepostingTask(pool(), checker
);
793 // Shutting down the pool should destroy the DestructionDeadlockCheckers,
794 // which in turn should not deadlock in their destructors.
798 // Verify that FlushForTesting works as intended.
799 TEST_F(SequencedWorkerPoolTest
, FlushForTesting
) {
800 // Should be fine to call on a new instance.
801 pool()->FlushForTesting();
803 // Queue up a bunch of work, including a long delayed task and
804 // a task that produces additional tasks as an artifact.
805 pool()->PostDelayedWorkerTask(
807 base::Bind(&TestTracker::FastTask
, tracker(), 0),
808 TimeDelta::FromMinutes(5));
809 pool()->PostWorkerTask(FROM_HERE
,
810 base::Bind(&TestTracker::SlowTask
, tracker(), 0));
811 const size_t kNumFastTasks
= 20;
812 for (size_t i
= 0; i
< kNumFastTasks
; i
++) {
813 pool()->PostWorkerTask(FROM_HERE
,
814 base::Bind(&TestTracker::FastTask
, tracker(), 0));
816 pool()->PostWorkerTask(
818 base::Bind(&TestTracker::PostAdditionalTasks
, tracker(), 0, pool(),
821 // We expect all except the delayed task to have been run. We verify all
822 // closures have been deleted by looking at the refcount of the
824 EXPECT_FALSE(tracker()->HasOneRef());
825 pool()->FlushForTesting();
826 EXPECT_TRUE(tracker()->HasOneRef());
827 EXPECT_EQ(1 + kNumFastTasks
+ 1 + 3, tracker()->GetTasksCompletedCount());
829 // Should be fine to call on an idle instance with all threads created, and
830 // spamming the method shouldn't deadlock or confuse the class.
831 pool()->FlushForTesting();
832 pool()->FlushForTesting();
834 // Should be fine to call after shutdown too.
836 pool()->FlushForTesting();
839 TEST(SequencedWorkerPoolRefPtrTest
, ShutsDownCleanWithContinueOnShutdown
) {
841 scoped_refptr
<SequencedWorkerPool
> pool(new SequencedWorkerPool(3, "Pool"));
842 scoped_refptr
<SequencedTaskRunner
> task_runner
=
843 pool
->GetSequencedTaskRunnerWithShutdownBehavior(
844 pool
->GetSequenceToken(),
845 base::SequencedWorkerPool::CONTINUE_ON_SHUTDOWN
);
847 // Upon test exit, should shut down without hanging.
851 class SequencedWorkerPoolTaskRunnerTestDelegate
{
853 SequencedWorkerPoolTaskRunnerTestDelegate() {}
855 ~SequencedWorkerPoolTaskRunnerTestDelegate() {}
857 void StartTaskRunner() {
859 new SequencedWorkerPoolOwner(10, "SequencedWorkerPoolTaskRunnerTest"));
862 scoped_refptr
<SequencedWorkerPool
> GetTaskRunner() {
863 return pool_owner_
->pool();
866 void StopTaskRunner() {
867 // Make sure all tasks are run before shutting down. Delayed tasks are
868 // not run, they're simply deleted.
869 pool_owner_
->pool()->FlushForTesting();
870 pool_owner_
->pool()->Shutdown();
871 // Don't reset |pool_owner_| here, as the test may still hold a
872 // reference to the pool.
876 MessageLoop message_loop_
;
877 scoped_ptr
<SequencedWorkerPoolOwner
> pool_owner_
;
880 INSTANTIATE_TYPED_TEST_CASE_P(
881 SequencedWorkerPool
, TaskRunnerTest
,
882 SequencedWorkerPoolTaskRunnerTestDelegate
);
884 class SequencedWorkerPoolTaskRunnerWithShutdownBehaviorTestDelegate
{
886 SequencedWorkerPoolTaskRunnerWithShutdownBehaviorTestDelegate() {}
888 ~SequencedWorkerPoolTaskRunnerWithShutdownBehaviorTestDelegate() {
891 void StartTaskRunner() {
893 new SequencedWorkerPoolOwner(10, "SequencedWorkerPoolTaskRunnerTest"));
894 task_runner_
= pool_owner_
->pool()->GetTaskRunnerWithShutdownBehavior(
895 SequencedWorkerPool::BLOCK_SHUTDOWN
);
898 scoped_refptr
<TaskRunner
> GetTaskRunner() {
902 void StopTaskRunner() {
903 // Make sure all tasks are run before shutting down. Delayed tasks are
904 // not run, they're simply deleted.
905 pool_owner_
->pool()->FlushForTesting();
906 pool_owner_
->pool()->Shutdown();
907 // Don't reset |pool_owner_| here, as the test may still hold a
908 // reference to the pool.
912 MessageLoop message_loop_
;
913 scoped_ptr
<SequencedWorkerPoolOwner
> pool_owner_
;
914 scoped_refptr
<TaskRunner
> task_runner_
;
917 INSTANTIATE_TYPED_TEST_CASE_P(
918 SequencedWorkerPoolTaskRunner
, TaskRunnerTest
,
919 SequencedWorkerPoolTaskRunnerWithShutdownBehaviorTestDelegate
);
921 class SequencedWorkerPoolSequencedTaskRunnerTestDelegate
{
923 SequencedWorkerPoolSequencedTaskRunnerTestDelegate() {}
925 ~SequencedWorkerPoolSequencedTaskRunnerTestDelegate() {
928 void StartTaskRunner() {
929 pool_owner_
.reset(new SequencedWorkerPoolOwner(
930 10, "SequencedWorkerPoolSequencedTaskRunnerTest"));
931 task_runner_
= pool_owner_
->pool()->GetSequencedTaskRunner(
932 pool_owner_
->pool()->GetSequenceToken());
935 scoped_refptr
<SequencedTaskRunner
> GetTaskRunner() {
939 void StopTaskRunner() {
940 // Make sure all tasks are run before shutting down. Delayed tasks are
941 // not run, they're simply deleted.
942 pool_owner_
->pool()->FlushForTesting();
943 pool_owner_
->pool()->Shutdown();
944 // Don't reset |pool_owner_| here, as the test may still hold a
945 // reference to the pool.
949 MessageLoop message_loop_
;
950 scoped_ptr
<SequencedWorkerPoolOwner
> pool_owner_
;
951 scoped_refptr
<SequencedTaskRunner
> task_runner_
;
954 INSTANTIATE_TYPED_TEST_CASE_P(
955 SequencedWorkerPoolSequencedTaskRunner
, TaskRunnerTest
,
956 SequencedWorkerPoolSequencedTaskRunnerTestDelegate
);
958 INSTANTIATE_TYPED_TEST_CASE_P(
959 SequencedWorkerPoolSequencedTaskRunner
, SequencedTaskRunnerTest
,
960 SequencedWorkerPoolSequencedTaskRunnerTestDelegate
);