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