base/threading/sequenced_worker_pool_unittest.cc

   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.
   4
   5 #include "base/threading/sequenced_worker_pool.h"
   6
   7 #include <algorithm>
   8
   9 #include "base/bind.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.h"
  14 #include "base/message_loop_proxy.h"
  15 #include "base/synchronization/condition_variable.h"
  16 #include "base/synchronization/lock.h"
  17 #include "base/test/sequenced_worker_pool_owner.h"
  18 #include "base/test/sequenced_task_runner_test_template.h"
  19 #include "base/test/task_runner_test_template.h"
  20 #include "base/threading/platform_thread.h"
  21 #include "base/tracked_objects.h"
  22 #include "testing/gtest/include/gtest/gtest.h"
  23
  24 namespace base {
  25
  26 // IMPORTANT NOTE:
  27 //
  28 // Many of these tests have failure modes where they'll hang forever. These
  29 // tests should not be flaky, and hangling indicates a type of failure. Do not
  30 // mark as flaky if they're hanging, it's likely an actual bug.
  31
  32 namespace {
  33
  34 const size_t kNumWorkerThreads = 3;
  35
  36 // Allows a number of threads to all be blocked on the same event, and
  37 // provides a way to unblock a certain number of them.
  38 class ThreadBlocker {
  39  public:
  40   ThreadBlocker() : lock_(), cond_var_(&lock_), unblock_counter_(0) {}
  41
  42   void Block() {
  43     {
  44       base::AutoLock lock(lock_);
  45       while (unblock_counter_ == 0)
  46         cond_var_.Wait();
  47       unblock_counter_--;
  48     }
  49     cond_var_.Signal();
  50   }
  51
  52   void Unblock(size_t count) {
  53     {
  54       base::AutoLock lock(lock_);
  55       DCHECK(unblock_counter_ == 0);
  56       unblock_counter_ = count;
  57     }
  58     cond_var_.Signal();
  59   }
  60
  61  private:
  62   base::Lock lock_;
  63   base::ConditionVariable cond_var_;
  64
  65   size_t unblock_counter_;
  66 };
  67
  68 class TestTracker : public base::RefCountedThreadSafe<TestTracker> {
  69  public:
  70   TestTracker()
  71       : lock_(),
  72         cond_var_(&lock_),
  73         started_events_(0) {
  74   }
  75
  76   // Each of these tasks appends the argument to the complete sequence vector
  77   // so calling code can see what order they finished in.
  78   void FastTask(int id) {
  79     SignalWorkerDone(id);
  80   }
  81
  82   void SlowTask(int id) {
  83     base::PlatformThread::Sleep(base::TimeDelta::FromSeconds(1));
  84     SignalWorkerDone(id);
  85   }
  86
  87   void BlockTask(int id, ThreadBlocker* blocker) {
  88     // Note that this task has started and signal anybody waiting for that
  89     // to happen.
  90     {
  91       base::AutoLock lock(lock_);
  92       started_events_++;
  93     }
  94     cond_var_.Signal();
  95
  96     blocker->Block();
  97     SignalWorkerDone(id);
  98   }
  99
 100   void PostAdditionalTasks(int id, SequencedWorkerPool* pool) {
 101     Closure fast_task = base::Bind(&TestTracker::FastTask, this, 100);
 102     EXPECT_FALSE(
 103         pool->PostWorkerTaskWithShutdownBehavior(
 104             FROM_HERE, fast_task,
 105             SequencedWorkerPool::CONTINUE_ON_SHUTDOWN));
 106     EXPECT_FALSE(
 107         pool->PostWorkerTaskWithShutdownBehavior(
 108             FROM_HERE, fast_task,
 109             SequencedWorkerPool::SKIP_ON_SHUTDOWN));
 110     pool->PostWorkerTaskWithShutdownBehavior(
 111         FROM_HERE, fast_task,
 112         SequencedWorkerPool::BLOCK_SHUTDOWN);
 113     SignalWorkerDone(id);
 114   }
 115
 116   // Waits until the given number of tasks have started executing.
 117   void WaitUntilTasksBlocked(size_t count) {
 118     {
 119       base::AutoLock lock(lock_);
 120       while (started_events_ < count)
 121         cond_var_.Wait();
 122     }
 123     cond_var_.Signal();
 124   }
 125
 126   // Blocks the current thread until at least the given number of tasks are in
 127   // the completed vector, and then returns a copy.
 128   std::vector<int> WaitUntilTasksComplete(size_t num_tasks) {
 129     std::vector<int> ret;
 130     {
 131       base::AutoLock lock(lock_);
 132       while (complete_sequence_.size() < num_tasks)
 133         cond_var_.Wait();
 134       ret = complete_sequence_;
 135     }
 136     cond_var_.Signal();
 137     return ret;
 138   }
 139
 140   void ClearCompleteSequence() {
 141     base::AutoLock lock(lock_);
 142     complete_sequence_.clear();
 143     started_events_ = 0;
 144   }
 145
 146  private:
 147   friend class base::RefCountedThreadSafe<TestTracker>;
 148   ~TestTracker() {}
 149
 150   void SignalWorkerDone(int id) {
 151     {
 152       base::AutoLock lock(lock_);
 153       complete_sequence_.push_back(id);
 154     }
 155     cond_var_.Signal();
 156   }
 157
 158   // Protects the complete_sequence.
 159   base::Lock lock_;
 160
 161   base::ConditionVariable cond_var_;
 162
 163   // Protected by lock_.
 164   std::vector<int> complete_sequence_;
 165
 166   // Counter of the number of "block" workers that have started.
 167   size_t started_events_;
 168 };
 169
 170 class SequencedWorkerPoolTest : public testing::Test {
 171  public:
 172   SequencedWorkerPoolTest()
 173       : pool_owner_(kNumWorkerThreads, "test"),
 174         tracker_(new TestTracker) {
 175   }
 176
 177   virtual ~SequencedWorkerPoolTest() {}
 178
 179   virtual void SetUp() OVERRIDE {}
 180
 181   virtual void TearDown() OVERRIDE {
 182     pool()->Shutdown();
 183   }
 184
 185   const scoped_refptr<SequencedWorkerPool>& pool() {
 186     return pool_owner_.pool();
 187   }
 188   TestTracker* tracker() { return tracker_.get(); }
 189
 190   void SetWillWaitForShutdownCallback(const Closure& callback) {
 191     pool_owner_.SetWillWaitForShutdownCallback(callback);
 192   }
 193
 194   // Ensures that the given number of worker threads is created by adding
 195   // tasks and waiting until they complete. Worker thread creation is
 196   // serialized, can happen on background threads asynchronously, and doesn't
 197   // happen any more at shutdown. This means that if a test posts a bunch of
 198   // tasks and calls shutdown, fewer workers will be created than the test may
 199   // expect.
 200   //
 201   // This function ensures that this condition can't happen so tests can make
 202   // assumptions about the number of workers active. See the comment in
 203   // PrepareToStartAdditionalThreadIfNecessary in the .cc file for more
 204   // details.
 205   //
 206   // It will post tasks to the queue with id -1. It also assumes this is the
 207   // first thing called in a test since it will clear the complete_sequence_.
 208   void EnsureAllWorkersCreated() {
 209     // Create a bunch of threads, all waiting. This will cause that may
 210     // workers to be created.
 211     ThreadBlocker blocker;
 212     for (size_t i = 0; i < kNumWorkerThreads; i++) {
 213       pool()->PostWorkerTask(FROM_HERE,
 214                              base::Bind(&TestTracker::BlockTask,
 215                                         tracker(), -1, &blocker));
 216     }
 217     tracker()->WaitUntilTasksBlocked(kNumWorkerThreads);
 218
 219     // Now wake them up and wait until they're done.
 220     blocker.Unblock(kNumWorkerThreads);
 221     tracker()->WaitUntilTasksComplete(kNumWorkerThreads);
 222
 223     // Clean up the task IDs we added.
 224     tracker()->ClearCompleteSequence();
 225   }
 226
 227   int has_work_call_count() const {
 228     return pool_owner_.has_work_call_count();
 229   }
 230
 231  private:
 232   MessageLoop message_loop_;
 233   SequencedWorkerPoolOwner pool_owner_;
 234   const scoped_refptr<TestTracker> tracker_;
 235 };
 236
 237 // Checks that the given number of entries are in the tasks to complete of
 238 // the given tracker, and then signals the given event the given number of
 239 // times. This is used to wakt up blocked background threads before blocking
 240 // on shutdown.
 241 void EnsureTasksToCompleteCountAndUnblock(scoped_refptr<TestTracker> tracker,
 242                                           size_t expected_tasks_to_complete,
 243                                           ThreadBlocker* blocker,
 244                                           size_t threads_to_awake) {
 245   EXPECT_EQ(
 246       expected_tasks_to_complete,
 247       tracker->WaitUntilTasksComplete(expected_tasks_to_complete).size());
 248
 249   blocker->Unblock(threads_to_awake);
 250 }
 251
 252 // Tests that same-named tokens have the same ID.
 253 TEST_F(SequencedWorkerPoolTest, NamedTokens) {
 254   const std::string name1("hello");
 255   SequencedWorkerPool::SequenceToken token1 =
 256       pool()->GetNamedSequenceToken(name1);
 257
 258   SequencedWorkerPool::SequenceToken token2 = pool()->GetSequenceToken();
 259
 260   const std::string name3("goodbye");
 261   SequencedWorkerPool::SequenceToken token3 =
 262       pool()->GetNamedSequenceToken(name3);
 263
 264   // All 3 tokens should be different.
 265   EXPECT_FALSE(token1.Equals(token2));
 266   EXPECT_FALSE(token1.Equals(token3));
 267   EXPECT_FALSE(token2.Equals(token3));
 268
 269   // Requesting the same name again should give the same value.
 270   SequencedWorkerPool::SequenceToken token1again =
 271       pool()->GetNamedSequenceToken(name1);
 272   EXPECT_TRUE(token1.Equals(token1again));
 273
 274   SequencedWorkerPool::SequenceToken token3again =
 275       pool()->GetNamedSequenceToken(name3);
 276   EXPECT_TRUE(token3.Equals(token3again));
 277 }
 278
 279 // Tests that posting a bunch of tasks (many more than the number of worker
 280 // threads) runs them all.
 281 TEST_F(SequencedWorkerPoolTest, LotsOfTasks) {
 282   pool()->PostWorkerTask(FROM_HERE,
 283                          base::Bind(&TestTracker::SlowTask, tracker(), 0));
 284
 285   const size_t kNumTasks = 20;
 286   for (size_t i = 1; i < kNumTasks; i++) {
 287     pool()->PostWorkerTask(FROM_HERE,
 288                            base::Bind(&TestTracker::FastTask, tracker(), i));
 289   }
 290
 291   std::vector<int> result = tracker()->WaitUntilTasksComplete(kNumTasks);
 292   EXPECT_EQ(kNumTasks, result.size());
 293 }
 294
 295 // Tests that posting a bunch of tasks (many more than the number of
 296 // worker threads) to two pools simultaneously runs them all twice.
 297 // This test is meant to shake out any concurrency issues between
 298 // pools (like histograms).
 299 TEST_F(SequencedWorkerPoolTest, LotsOfTasksTwoPools) {
 300   SequencedWorkerPoolOwner pool1(kNumWorkerThreads, "test1");
 301   SequencedWorkerPoolOwner pool2(kNumWorkerThreads, "test2");
 302
 303   base::Closure slow_task = base::Bind(&TestTracker::SlowTask, tracker(), 0);
 304   pool1.pool()->PostWorkerTask(FROM_HERE, slow_task);
 305   pool2.pool()->PostWorkerTask(FROM_HERE, slow_task);
 306
 307   const size_t kNumTasks = 20;
 308   for (size_t i = 1; i < kNumTasks; i++) {
 309     base::Closure fast_task =
 310         base::Bind(&TestTracker::FastTask, tracker(), i);
 311     pool1.pool()->PostWorkerTask(FROM_HERE, fast_task);
 312     pool2.pool()->PostWorkerTask(FROM_HERE, fast_task);
 313   }
 314
 315   std::vector<int> result =
 316       tracker()->WaitUntilTasksComplete(2*kNumTasks);
 317   EXPECT_EQ(2 * kNumTasks, result.size());
 318
 319   pool2.pool()->Shutdown();
 320   pool1.pool()->Shutdown();
 321 }
 322
 323 // Test that tasks with the same sequence token are executed in order but don't
 324 // affect other tasks.
 325 TEST_F(SequencedWorkerPoolTest, Sequence) {
 326   // Fill all the worker threads except one.
 327   const size_t kNumBackgroundTasks = kNumWorkerThreads - 1;
 328   ThreadBlocker background_blocker;
 329   for (size_t i = 0; i < kNumBackgroundTasks; i++) {
 330     pool()->PostWorkerTask(FROM_HERE,
 331                            base::Bind(&TestTracker::BlockTask,
 332                                       tracker(), i, &background_blocker));
 333   }
 334   tracker()->WaitUntilTasksBlocked(kNumBackgroundTasks);
 335
 336   // Create two tasks with the same sequence token, one that will block on the
 337   // event, and one which will just complete quickly when it's run. Since there
 338   // is one worker thread free, the first task will start and then block, and
 339   // the second task should be waiting.
 340   ThreadBlocker blocker;
 341   SequencedWorkerPool::SequenceToken token1 = pool()->GetSequenceToken();
 342   pool()->PostSequencedWorkerTask(
 343       token1, FROM_HERE,
 344       base::Bind(&TestTracker::BlockTask, tracker(), 100, &blocker));
 345   pool()->PostSequencedWorkerTask(
 346       token1, FROM_HERE,
 347       base::Bind(&TestTracker::FastTask, tracker(), 101));
 348   EXPECT_EQ(0u, tracker()->WaitUntilTasksComplete(0).size());
 349
 350   // Create another two tasks as above with a different token. These will be
 351   // blocked since there are no slots to run.
 352   SequencedWorkerPool::SequenceToken token2 = pool()->GetSequenceToken();
 353   pool()->PostSequencedWorkerTask(
 354       token2, FROM_HERE,
 355       base::Bind(&TestTracker::FastTask, tracker(), 200));
 356   pool()->PostSequencedWorkerTask(
 357       token2, FROM_HERE,
 358       base::Bind(&TestTracker::FastTask, tracker(), 201));
 359   EXPECT_EQ(0u, tracker()->WaitUntilTasksComplete(0).size());
 360
 361   // Let one background task complete. This should then let both tasks of
 362   // token2 run to completion in order. The second task of token1 should still
 363   // be blocked.
 364   background_blocker.Unblock(1);
 365   std::vector<int> result = tracker()->WaitUntilTasksComplete(3);
 366   ASSERT_EQ(3u, result.size());
 367   EXPECT_EQ(200, result[1]);
 368   EXPECT_EQ(201, result[2]);
 369
 370   // Finish the rest of the background tasks. This should leave some workers
 371   // free with the second token1 task still blocked on the first.
 372   background_blocker.Unblock(kNumBackgroundTasks - 1);
 373   EXPECT_EQ(kNumBackgroundTasks + 2,
 374             tracker()->WaitUntilTasksComplete(kNumBackgroundTasks + 2).size());
 375
 376   // Allow the first task of token1 to complete. This should run the second.
 377   blocker.Unblock(1);
 378   result = tracker()->WaitUntilTasksComplete(kNumBackgroundTasks + 4);
 379   ASSERT_EQ(kNumBackgroundTasks + 4, result.size());
 380   EXPECT_EQ(100, result[result.size() - 2]);
 381   EXPECT_EQ(101, result[result.size() - 1]);
 382 }
 383
 384 // Tests that any tasks posted after Shutdown are ignored.
 385 TEST_F(SequencedWorkerPoolTest, IgnoresAfterShutdown) {
 386   // Start tasks to take all the threads and block them.
 387   EnsureAllWorkersCreated();
 388   ThreadBlocker blocker;
 389   for (size_t i = 0; i < kNumWorkerThreads; i++) {
 390     pool()->PostWorkerTask(FROM_HERE,
 391                            base::Bind(&TestTracker::BlockTask,
 392                                       tracker(), i, &blocker));
 393   }
 394   tracker()->WaitUntilTasksBlocked(kNumWorkerThreads);
 395
 396   SetWillWaitForShutdownCallback(
 397       base::Bind(&EnsureTasksToCompleteCountAndUnblock,
 398                  scoped_refptr<TestTracker>(tracker()), 0,
 399                  &blocker, kNumWorkerThreads));
 400
 401   // Shutdown the worker pool. This should discard all non-blocking tasks.
 402   const int kMaxNewBlockingTasksAfterShutdown = 100;
 403   pool()->Shutdown(kMaxNewBlockingTasksAfterShutdown);
 404
 405   int old_has_work_call_count = has_work_call_count();
 406
 407   std::vector<int> result =
 408       tracker()->WaitUntilTasksComplete(kNumWorkerThreads);
 409
 410   // The kNumWorkerThread items should have completed, in no particular order.
 411   ASSERT_EQ(kNumWorkerThreads, result.size());
 412   for (size_t i = 0; i < kNumWorkerThreads; i++) {
 413     EXPECT_TRUE(std::find(result.begin(), result.end(), static_cast<int>(i)) !=
 414                 result.end());
 415   }
 416
 417   // No further tasks, regardless of shutdown mode, should be allowed.
 418   EXPECT_FALSE(pool()->PostWorkerTaskWithShutdownBehavior(
 419       FROM_HERE,
 420       base::Bind(&TestTracker::FastTask, tracker(), 100),
 421       SequencedWorkerPool::CONTINUE_ON_SHUTDOWN));
 422   EXPECT_FALSE(pool()->PostWorkerTaskWithShutdownBehavior(
 423       FROM_HERE,
 424       base::Bind(&TestTracker::FastTask, tracker(), 101),
 425       SequencedWorkerPool::SKIP_ON_SHUTDOWN));
 426   EXPECT_FALSE(pool()->PostWorkerTaskWithShutdownBehavior(
 427       FROM_HERE,
 428       base::Bind(&TestTracker::FastTask, tracker(), 102),
 429       SequencedWorkerPool::BLOCK_SHUTDOWN));
 430
 431   ASSERT_EQ(old_has_work_call_count, has_work_call_count());
 432 }
 433
 434 TEST_F(SequencedWorkerPoolTest, AllowsAfterShutdown) {
 435   // Test that <n> new blocking tasks are allowed provided they're posted
 436   // by a running tasks.
 437   EnsureAllWorkersCreated();
 438   ThreadBlocker blocker;
 439
 440   // Start tasks to take all the threads and block them.
 441   const int kNumBlockTasks = static_cast<int>(kNumWorkerThreads);
 442   for (int i = 0; i < kNumBlockTasks; ++i) {
 443     EXPECT_TRUE(pool()->PostWorkerTask(
 444         FROM_HERE,
 445         base::Bind(&TestTracker::BlockTask, tracker(), i, &blocker)));
 446   }
 447   tracker()->WaitUntilTasksBlocked(kNumWorkerThreads);
 448
 449   // Queue up shutdown blocking tasks behind those which will attempt to post
 450   // additional tasks when run, PostAdditionalTasks attemtps to post 3
 451   // new FastTasks, one for each shutdown_behavior.
 452   const int kNumQueuedTasks = static_cast<int>(kNumWorkerThreads);
 453   for (int i = 0; i < kNumQueuedTasks; ++i) {
 454     EXPECT_TRUE(pool()->PostWorkerTaskWithShutdownBehavior(
 455         FROM_HERE,
 456         base::Bind(&TestTracker::PostAdditionalTasks, tracker(), i, pool()),
 457         SequencedWorkerPool::BLOCK_SHUTDOWN));
 458   }
 459
 460   // Setup to open the floodgates from within Shutdown().
 461   SetWillWaitForShutdownCallback(
 462       base::Bind(&EnsureTasksToCompleteCountAndUnblock,
 463                  scoped_refptr<TestTracker>(tracker()),
 464                  0, &blocker, kNumBlockTasks));
 465
 466   // Allow half of the additional blocking tasks thru.
 467   const int kNumNewBlockingTasksToAllow = kNumWorkerThreads / 2;
 468   pool()->Shutdown(kNumNewBlockingTasksToAllow);
 469
 470   // Ensure that the correct number of tasks actually got run.
 471   tracker()->WaitUntilTasksComplete(static_cast<size_t>(
 472       kNumBlockTasks + kNumQueuedTasks + kNumNewBlockingTasksToAllow));
 473
 474   // Clean up the task IDs we added and go home.
 475   tracker()->ClearCompleteSequence();
 476 }
 477
 478 // Tests that unrun tasks are discarded properly according to their shutdown
 479 // mode.
 480 TEST_F(SequencedWorkerPoolTest, DiscardOnShutdown) {
 481   // Start tasks to take all the threads and block them.
 482   EnsureAllWorkersCreated();
 483   ThreadBlocker blocker;
 484   for (size_t i = 0; i < kNumWorkerThreads; i++) {
 485     pool()->PostWorkerTask(FROM_HERE,
 486                            base::Bind(&TestTracker::BlockTask,
 487                                       tracker(), i, &blocker));
 488   }
 489   tracker()->WaitUntilTasksBlocked(kNumWorkerThreads);
 490
 491   // Create some tasks with different shutdown modes.
 492   pool()->PostWorkerTaskWithShutdownBehavior(
 493       FROM_HERE,
 494       base::Bind(&TestTracker::FastTask, tracker(), 100),
 495       SequencedWorkerPool::CONTINUE_ON_SHUTDOWN);
 496   pool()->PostWorkerTaskWithShutdownBehavior(
 497       FROM_HERE,
 498       base::Bind(&TestTracker::FastTask, tracker(), 101),
 499       SequencedWorkerPool::SKIP_ON_SHUTDOWN);
 500   pool()->PostWorkerTaskWithShutdownBehavior(
 501       FROM_HERE,
 502       base::Bind(&TestTracker::FastTask, tracker(), 102),
 503       SequencedWorkerPool::BLOCK_SHUTDOWN);
 504
 505   // Shutdown the worker pool. This should discard all non-blocking tasks.
 506   SetWillWaitForShutdownCallback(
 507       base::Bind(&EnsureTasksToCompleteCountAndUnblock,
 508                  scoped_refptr<TestTracker>(tracker()), 0,
 509                  &blocker, kNumWorkerThreads));
 510   pool()->Shutdown();
 511
 512   std::vector<int> result =
 513       tracker()->WaitUntilTasksComplete(kNumWorkerThreads + 1);
 514
 515   // The kNumWorkerThread items should have completed, plus the BLOCK_SHUTDOWN
 516   // one, in no particular order.
 517   ASSERT_EQ(kNumWorkerThreads + 1, result.size());
 518   for (size_t i = 0; i < kNumWorkerThreads; i++) {
 519     EXPECT_TRUE(std::find(result.begin(), result.end(), static_cast<int>(i)) !=
 520                 result.end());
 521   }
 522   EXPECT_TRUE(std::find(result.begin(), result.end(), 102) != result.end());
 523 }
 524
 525 // Tests that CONTINUE_ON_SHUTDOWN tasks don't block shutdown.
 526 TEST_F(SequencedWorkerPoolTest, ContinueOnShutdown) {
 527   scoped_refptr<TaskRunner> runner(pool()->GetTaskRunnerWithShutdownBehavior(
 528       SequencedWorkerPool::CONTINUE_ON_SHUTDOWN));
 529   scoped_refptr<SequencedTaskRunner> sequenced_runner(
 530       pool()->GetSequencedTaskRunnerWithShutdownBehavior(
 531           pool()->GetSequenceToken(),
 532           SequencedWorkerPool::CONTINUE_ON_SHUTDOWN));
 533   EnsureAllWorkersCreated();
 534   ThreadBlocker blocker;
 535   pool()->PostWorkerTaskWithShutdownBehavior(
 536       FROM_HERE,
 537       base::Bind(&TestTracker::BlockTask,
 538                  tracker(), 0, &blocker),
 539       SequencedWorkerPool::CONTINUE_ON_SHUTDOWN);
 540   runner->PostTask(
 541       FROM_HERE,
 542       base::Bind(&TestTracker::BlockTask,
 543                  tracker(), 1, &blocker));
 544   sequenced_runner->PostTask(
 545       FROM_HERE,
 546       base::Bind(&TestTracker::BlockTask,
 547                  tracker(), 2, &blocker));
 548
 549   tracker()->WaitUntilTasksBlocked(3);
 550
 551   // This should not block. If this test hangs, it means it failed.
 552   pool()->Shutdown();
 553
 554   // The task should not have completed yet.
 555   EXPECT_EQ(0u, tracker()->WaitUntilTasksComplete(0).size());
 556
 557   // Posting more tasks should fail.
 558   EXPECT_FALSE(pool()->PostWorkerTaskWithShutdownBehavior(
 559       FROM_HERE, base::Bind(&TestTracker::FastTask, tracker(), 0),
 560       SequencedWorkerPool::CONTINUE_ON_SHUTDOWN));
 561   EXPECT_FALSE(runner->PostTask(
 562       FROM_HERE, base::Bind(&TestTracker::FastTask, tracker(), 0)));
 563   EXPECT_FALSE(sequenced_runner->PostTask(
 564       FROM_HERE, base::Bind(&TestTracker::FastTask, tracker(), 0)));
 565
 566   // Continue the background thread and make sure the tasks can complete.
 567   blocker.Unblock(3);
 568   std::vector<int> result = tracker()->WaitUntilTasksComplete(3);
 569   EXPECT_EQ(3u, result.size());
 570 }
 571
 572 // Tests that SKIP_ON_SHUTDOWN tasks that have been started block Shutdown
 573 // until they stop, but tasks not yet started do not.
 574 TEST_F(SequencedWorkerPoolTest, SkipOnShutdown) {
 575   // Start tasks to take all the threads and block them.
 576   EnsureAllWorkersCreated();
 577   ThreadBlocker blocker;
 578
 579   // Now block all the threads with SKIP_ON_SHUTDOWN. Shutdown() should not
 580   // return until these tasks have completed.
 581   for (size_t i = 0; i < kNumWorkerThreads; i++) {
 582     pool()->PostWorkerTaskWithShutdownBehavior(
 583         FROM_HERE,
 584         base::Bind(&TestTracker::BlockTask, tracker(), i, &blocker),
 585         SequencedWorkerPool::SKIP_ON_SHUTDOWN);
 586   }
 587   tracker()->WaitUntilTasksBlocked(kNumWorkerThreads);
 588
 589   // Now post an additional task as SKIP_ON_SHUTDOWN, which should not be
 590   // executed once Shutdown() has been called.
 591   pool()->PostWorkerTaskWithShutdownBehavior(
 592       FROM_HERE,
 593       base::Bind(&TestTracker::BlockTask,
 594                  tracker(), 0, &blocker),
 595       SequencedWorkerPool::SKIP_ON_SHUTDOWN);
 596
 597   // This callback will only be invoked if SKIP_ON_SHUTDOWN tasks that have
 598   // been started block shutdown.
 599   SetWillWaitForShutdownCallback(
 600       base::Bind(&EnsureTasksToCompleteCountAndUnblock,
 601                  scoped_refptr<TestTracker>(tracker()), 0,
 602                  &blocker, kNumWorkerThreads));
 603
 604   // No tasks should have completed yet.
 605   EXPECT_EQ(0u, tracker()->WaitUntilTasksComplete(0).size());
 606
 607   // This should not block. If this test hangs, it means it failed.
 608   pool()->Shutdown();
 609
 610   // Shutdown should not return until all of the tasks have completed.
 611   std::vector<int> result =
 612       tracker()->WaitUntilTasksComplete(kNumWorkerThreads);
 613
 614   // Only tasks marked SKIP_ON_SHUTDOWN that were already started should be
 615   // allowed to complete. No additional non-blocking tasks should have been
 616   // started.
 617   ASSERT_EQ(kNumWorkerThreads, result.size());
 618   for (size_t i = 0; i < kNumWorkerThreads; i++) {
 619     EXPECT_TRUE(std::find(result.begin(), result.end(), static_cast<int>(i)) !=
 620                 result.end());
 621   }
 622 }
 623
 624 // Ensure all worker threads are created, and then trigger a spurious
 625 // work signal. This shouldn't cause any other work signals to be
 626 // triggered. This is a regression test for http://crbug.com/117469.
 627 TEST_F(SequencedWorkerPoolTest, SpuriousWorkSignal) {
 628   EnsureAllWorkersCreated();
 629   int old_has_work_call_count = has_work_call_count();
 630   pool()->SignalHasWorkForTesting();
 631   // This is inherently racy, but can only produce false positives.
 632   base::PlatformThread::Sleep(base::TimeDelta::FromMilliseconds(100));
 633   EXPECT_EQ(old_has_work_call_count + 1, has_work_call_count());
 634 }
 635
 636 void IsRunningOnCurrentThreadTask(
 637     SequencedWorkerPool::SequenceToken test_positive_token,
 638     SequencedWorkerPool::SequenceToken test_negative_token,
 639     SequencedWorkerPool* pool,
 640     SequencedWorkerPool* unused_pool) {
 641   EXPECT_TRUE(pool->RunsTasksOnCurrentThread());
 642   EXPECT_TRUE(pool->IsRunningSequenceOnCurrentThread(test_positive_token));
 643   EXPECT_FALSE(pool->IsRunningSequenceOnCurrentThread(test_negative_token));
 644   EXPECT_FALSE(unused_pool->RunsTasksOnCurrentThread());
 645   EXPECT_FALSE(
 646       unused_pool->IsRunningSequenceOnCurrentThread(test_positive_token));
 647   EXPECT_FALSE(
 648       unused_pool->IsRunningSequenceOnCurrentThread(test_negative_token));
 649 }
 650
 651 // Verify correctness of the IsRunningSequenceOnCurrentThread method.
 652 TEST_F(SequencedWorkerPoolTest, IsRunningOnCurrentThread) {
 653   SequencedWorkerPool::SequenceToken token1 = pool()->GetSequenceToken();
 654   SequencedWorkerPool::SequenceToken token2 = pool()->GetSequenceToken();
 655   SequencedWorkerPool::SequenceToken unsequenced_token;
 656
 657   scoped_refptr<SequencedWorkerPool> unused_pool =
 658       new SequencedWorkerPool(2, "unused_pool");
 659   EXPECT_TRUE(token1.Equals(unused_pool->GetSequenceToken()));
 660   EXPECT_TRUE(token2.Equals(unused_pool->GetSequenceToken()));
 661
 662   EXPECT_FALSE(pool()->RunsTasksOnCurrentThread());
 663   EXPECT_FALSE(pool()->IsRunningSequenceOnCurrentThread(token1));
 664   EXPECT_FALSE(pool()->IsRunningSequenceOnCurrentThread(token2));
 665   EXPECT_FALSE(pool()->IsRunningSequenceOnCurrentThread(unsequenced_token));
 666   EXPECT_FALSE(unused_pool->RunsTasksOnCurrentThread());
 667   EXPECT_FALSE(unused_pool->IsRunningSequenceOnCurrentThread(token1));
 668   EXPECT_FALSE(unused_pool->IsRunningSequenceOnCurrentThread(token2));
 669   EXPECT_FALSE(
 670       unused_pool->IsRunningSequenceOnCurrentThread(unsequenced_token));
 671
 672   pool()->PostSequencedWorkerTask(
 673       token1, FROM_HERE,
 674       base::Bind(&IsRunningOnCurrentThreadTask,
 675                  token1, token2, pool(), unused_pool));
 676   pool()->PostSequencedWorkerTask(
 677       token2, FROM_HERE,
 678       base::Bind(&IsRunningOnCurrentThreadTask,
 679                  token2, unsequenced_token, pool(), unused_pool));
 680   pool()->PostWorkerTask(
 681       FROM_HERE,
 682       base::Bind(&IsRunningOnCurrentThreadTask,
 683                  unsequenced_token, token1, pool(), unused_pool));
 684   pool()->Shutdown();
 685   unused_pool->Shutdown();
 686 }
 687
 688 class SequencedWorkerPoolTaskRunnerTestDelegate {
 689  public:
 690   SequencedWorkerPoolTaskRunnerTestDelegate() {}
 691
 692   ~SequencedWorkerPoolTaskRunnerTestDelegate() {}
 693
 694   void StartTaskRunner() {
 695     pool_owner_.reset(
 696         new SequencedWorkerPoolOwner(10, "SequencedWorkerPoolTaskRunnerTest"));
 697   }
 698
 699   scoped_refptr<SequencedWorkerPool> GetTaskRunner() {
 700     return pool_owner_->pool();
 701   }
 702
 703   void StopTaskRunner() {
 704     // Make sure all tasks (including delayed ones) are run before shutting
 705     // down.
 706     pool_owner_->pool()->FlushForTesting();
 707     pool_owner_->pool()->Shutdown();
 708     // Don't reset |pool_owner_| here, as the test may still hold a
 709     // reference to the pool.
 710   }
 711
 712   bool TaskRunnerHandlesNonZeroDelays() const {
 713     return true;
 714   }
 715
 716  private:
 717   MessageLoop message_loop_;
 718   scoped_ptr<SequencedWorkerPoolOwner> pool_owner_;
 719 };
 720
 721 INSTANTIATE_TYPED_TEST_CASE_P(
 722     SequencedWorkerPool, TaskRunnerTest,
 723     SequencedWorkerPoolTaskRunnerTestDelegate);
 724
 725 class SequencedWorkerPoolTaskRunnerWithShutdownBehaviorTestDelegate {
 726  public:
 727   SequencedWorkerPoolTaskRunnerWithShutdownBehaviorTestDelegate() {}
 728
 729   ~SequencedWorkerPoolTaskRunnerWithShutdownBehaviorTestDelegate() {
 730   }
 731
 732   void StartTaskRunner() {
 733     pool_owner_.reset(
 734         new SequencedWorkerPoolOwner(10, "SequencedWorkerPoolTaskRunnerTest"));
 735     task_runner_ = pool_owner_->pool()->GetTaskRunnerWithShutdownBehavior(
 736         SequencedWorkerPool::BLOCK_SHUTDOWN);
 737   }
 738
 739   scoped_refptr<TaskRunner> GetTaskRunner() {
 740     return task_runner_;
 741   }
 742
 743   void StopTaskRunner() {
 744     // Make sure all tasks (including delayed ones) are run before shutting
 745     // down.
 746     pool_owner_->pool()->FlushForTesting();
 747     pool_owner_->pool()->Shutdown();
 748     // Don't reset |pool_owner_| here, as the test may still hold a
 749     // reference to the pool.
 750   }
 751
 752   bool TaskRunnerHandlesNonZeroDelays() const {
 753     return true;
 754   }
 755
 756  private:
 757   MessageLoop message_loop_;
 758   scoped_ptr<SequencedWorkerPoolOwner> pool_owner_;
 759   scoped_refptr<TaskRunner> task_runner_;
 760 };
 761
 762 INSTANTIATE_TYPED_TEST_CASE_P(
 763     SequencedWorkerPoolTaskRunner, TaskRunnerTest,
 764     SequencedWorkerPoolTaskRunnerWithShutdownBehaviorTestDelegate);
 765
 766 class SequencedWorkerPoolSequencedTaskRunnerTestDelegate {
 767  public:
 768   SequencedWorkerPoolSequencedTaskRunnerTestDelegate() {}
 769
 770   ~SequencedWorkerPoolSequencedTaskRunnerTestDelegate() {
 771   }
 772
 773   void StartTaskRunner() {
 774     pool_owner_.reset(new SequencedWorkerPoolOwner(
 775         10, "SequencedWorkerPoolSequencedTaskRunnerTest"));
 776     task_runner_ = pool_owner_->pool()->GetSequencedTaskRunner(
 777         pool_owner_->pool()->GetSequenceToken());
 778   }
 779
 780   scoped_refptr<SequencedTaskRunner> GetTaskRunner() {
 781     return task_runner_;
 782   }
 783
 784   void StopTaskRunner() {
 785     // Make sure all tasks (including delayed ones) are run before shutting
 786     // down.
 787     pool_owner_->pool()->FlushForTesting();
 788     pool_owner_->pool()->Shutdown();
 789     // Don't reset |pool_owner_| here, as the test may still hold a
 790     // reference to the pool.
 791   }
 792
 793   bool TaskRunnerHandlesNonZeroDelays() const {
 794     return true;
 795   }
 796
 797  private:
 798   MessageLoop message_loop_;
 799   scoped_ptr<SequencedWorkerPoolOwner> pool_owner_;
 800   scoped_refptr<SequencedTaskRunner> task_runner_;
 801 };
 802
 803 INSTANTIATE_TYPED_TEST_CASE_P(
 804     SequencedWorkerPoolSequencedTaskRunner, TaskRunnerTest,
 805     SequencedWorkerPoolSequencedTaskRunnerTestDelegate);
 806
 807 INSTANTIATE_TYPED_TEST_CASE_P(
 808     SequencedWorkerPoolSequencedTaskRunner, SequencedTaskRunnerTest,
 809     SequencedWorkerPoolSequencedTaskRunnerTestDelegate);
 810
 811 }  // namespace
 812
 813 }  // namespace base