base/synchronization/condition_variable_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 // Multi-threaded tests of ConditionVariable class.
   6
   7 #include <time.h>
   8 #include <algorithm>
   9 #include <vector>
  10
  11 #include "base/logging.h"
  12 #include "base/memory/scoped_ptr.h"
  13 #include "base/synchronization/condition_variable.h"
  14 #include "base/synchronization/lock.h"
  15 #include "base/synchronization/spin_wait.h"
  16 #include "base/threading/platform_thread.h"
  17 #include "base/threading/thread_collision_warner.h"
  18 #include "base/time.h"
  19 #include "testing/gtest/include/gtest/gtest.h"
  20 #include "testing/platform_test.h"
  21
  22 namespace base {
  23
  24 namespace {
  25 //------------------------------------------------------------------------------
  26 // Define our test class, with several common variables.
  27 //------------------------------------------------------------------------------
  28
  29 class ConditionVariableTest : public PlatformTest {
  30  public:
  31   const TimeDelta kZeroMs;
  32   const TimeDelta kTenMs;
  33   const TimeDelta kThirtyMs;
  34   const TimeDelta kFortyFiveMs;
  35   const TimeDelta kSixtyMs;
  36   const TimeDelta kOneHundredMs;
  37
  38   explicit ConditionVariableTest()
  39       : kZeroMs(TimeDelta::FromMilliseconds(0)),
  40         kTenMs(TimeDelta::FromMilliseconds(10)),
  41         kThirtyMs(TimeDelta::FromMilliseconds(30)),
  42         kFortyFiveMs(TimeDelta::FromMilliseconds(45)),
  43         kSixtyMs(TimeDelta::FromMilliseconds(60)),
  44         kOneHundredMs(TimeDelta::FromMilliseconds(100)) {
  45   }
  46 };
  47
  48 //------------------------------------------------------------------------------
  49 // Define a class that will control activities an several multi-threaded tests.
  50 // The general structure of multi-threaded tests is that a test case will
  51 // construct an instance of a WorkQueue.  The WorkQueue will spin up some
  52 // threads and control them throughout their lifetime, as well as maintaining
  53 // a central repository of the work thread's activity.  Finally, the WorkQueue
  54 // will command the the worker threads to terminate.  At that point, the test
  55 // cases will validate that the WorkQueue has records showing that the desired
  56 // activities were performed.
  57 //------------------------------------------------------------------------------
  58
  59 // Callers are responsible for synchronizing access to the following class.
  60 // The WorkQueue::lock_, as accessed via WorkQueue::lock(), should be used for
  61 // all synchronized access.
  62 class WorkQueue : public PlatformThread::Delegate {
  63  public:
  64   explicit WorkQueue(int thread_count);
  65   ~WorkQueue();
  66
  67   // PlatformThread::Delegate interface.
  68   void ThreadMain();
  69
  70   //----------------------------------------------------------------------------
  71   // Worker threads only call the following methods.
  72   // They should use the lock to get exclusive access.
  73   int GetThreadId();  // Get an ID assigned to a thread..
  74   bool EveryIdWasAllocated() const;  // Indicates that all IDs were handed out.
  75   TimeDelta GetAnAssignment(int thread_id);  // Get a work task duration.
  76   void WorkIsCompleted(int thread_id);
  77
  78   int task_count() const;
  79   bool allow_help_requests() const;  // Workers can signal more workers.
  80   bool shutdown() const;  // Check if shutdown has been requested.
  81
  82   void thread_shutting_down();
  83
  84
  85   //----------------------------------------------------------------------------
  86   // Worker threads can call them but not needed to acquire a lock.
  87   Lock* lock();
  88
  89   ConditionVariable* work_is_available();
  90   ConditionVariable* all_threads_have_ids();
  91   ConditionVariable* no_more_tasks();
  92
  93   //----------------------------------------------------------------------------
  94   // The rest of the methods are for use by the controlling master thread (the
  95   // test case code).
  96   void ResetHistory();
  97   int GetMinCompletionsByWorkerThread() const;
  98   int GetMaxCompletionsByWorkerThread() const;
  99   int GetNumThreadsTakingAssignments() const;
 100   int GetNumThreadsCompletingTasks() const;
 101   int GetNumberOfCompletedTasks() const;
 102   TimeDelta GetWorkTime() const;
 103
 104   void SetWorkTime(TimeDelta delay);
 105   void SetTaskCount(int count);
 106   void SetAllowHelp(bool allow);
 107
 108   // The following must be called without locking, and will spin wait until the
 109   // threads are all in a wait state.
 110   void SpinUntilAllThreadsAreWaiting();
 111   void SpinUntilTaskCountLessThan(int task_count);
 112
 113   // Caller must acquire lock before calling.
 114   void SetShutdown();
 115
 116   // Compares the |shutdown_task_count_| to the |thread_count| and returns true
 117   // if they are equal.  This check will acquire the |lock_| so the caller
 118   // should not hold the lock when calling this method.
 119   bool ThreadSafeCheckShutdown(int thread_count);
 120
 121  private:
 122   // Both worker threads and controller use the following to synchronize.
 123   Lock lock_;
 124   ConditionVariable work_is_available_;  // To tell threads there is work.
 125
 126   // Conditions to notify the controlling process (if it is interested).
 127   ConditionVariable all_threads_have_ids_;  // All threads are running.
 128   ConditionVariable no_more_tasks_;  // Task count is zero.
 129
 130   const int thread_count_;
 131   int waiting_thread_count_;
 132   scoped_array<PlatformThreadHandle> thread_handles_;
 133   std::vector<int> assignment_history_;  // Number of assignment per worker.
 134   std::vector<int> completion_history_;  // Number of completions per worker.
 135   int thread_started_counter_;  // Used to issue unique id to workers.
 136   int shutdown_task_count_;  // Number of tasks told to shutdown
 137   int task_count_;  // Number of assignment tasks waiting to be processed.
 138   TimeDelta worker_delay_;  // Time each task takes to complete.
 139   bool allow_help_requests_;  // Workers can signal more workers.
 140   bool shutdown_;  // Set when threads need to terminate.
 141
 142   DFAKE_MUTEX(locked_methods_);
 143 };
 144
 145 //------------------------------------------------------------------------------
 146 // The next section contains the actual tests.
 147 //------------------------------------------------------------------------------
 148
 149 TEST_F(ConditionVariableTest, StartupShutdownTest) {
 150   Lock lock;
 151
 152   // First try trivial startup/shutdown.
 153   {
 154     ConditionVariable cv1(&lock);
 155   }  // Call for cv1 destruction.
 156
 157   // Exercise with at least a few waits.
 158   ConditionVariable cv(&lock);
 159
 160   lock.Acquire();
 161   cv.TimedWait(kTenMs);  // Wait for 10 ms.
 162   cv.TimedWait(kTenMs);  // Wait for 10 ms.
 163   lock.Release();
 164
 165   lock.Acquire();
 166   cv.TimedWait(kTenMs);  // Wait for 10 ms.
 167   cv.TimedWait(kTenMs);  // Wait for 10 ms.
 168   cv.TimedWait(kTenMs);  // Wait for 10 ms.
 169   lock.Release();
 170 }  // Call for cv destruction.
 171
 172 TEST_F(ConditionVariableTest, TimeoutTest) {
 173   Lock lock;
 174   ConditionVariable cv(&lock);
 175   lock.Acquire();
 176
 177   TimeTicks start = TimeTicks::Now();
 178   const TimeDelta WAIT_TIME = TimeDelta::FromMilliseconds(300);
 179   // Allow for clocking rate granularity.
 180   const TimeDelta FUDGE_TIME = TimeDelta::FromMilliseconds(50);
 181
 182   cv.TimedWait(WAIT_TIME + FUDGE_TIME);
 183   TimeDelta duration = TimeTicks::Now() - start;
 184   // We can't use EXPECT_GE here as the TimeDelta class does not support the
 185   // required stream conversion.
 186   EXPECT_TRUE(duration >= WAIT_TIME);
 187
 188   lock.Release();
 189 }
 190
 191
 192 // Suddenly got flaky on Win, see http://crbug.com/10607 (starting at
 193 // comment #15)
 194 #if defined(OS_WIN)
 195 #define MAYBE_MultiThreadConsumerTest DISABLED_MultiThreadConsumerTest
 196 #else
 197 #define MAYBE_MultiThreadConsumerTest MultiThreadConsumerTest
 198 #endif
 199 // Test serial task servicing, as well as two parallel task servicing methods.
 200 TEST_F(ConditionVariableTest, MAYBE_MultiThreadConsumerTest) {
 201   const int kThreadCount = 10;
 202   WorkQueue queue(kThreadCount);  // Start the threads.
 203
 204   const int kTaskCount = 10;  // Number of tasks in each mini-test here.
 205
 206   Time start_time;  // Used to time task processing.
 207
 208   {
 209     base::AutoLock auto_lock(*queue.lock());
 210     while (!queue.EveryIdWasAllocated())
 211       queue.all_threads_have_ids()->Wait();
 212   }
 213
 214   // If threads aren't in a wait state, they may start to gobble up tasks in
 215   // parallel, short-circuiting (breaking) this test.
 216   queue.SpinUntilAllThreadsAreWaiting();
 217
 218   {
 219     // Since we have no tasks yet, all threads should be waiting by now.
 220     base::AutoLock auto_lock(*queue.lock());
 221     EXPECT_EQ(0, queue.GetNumThreadsTakingAssignments());
 222     EXPECT_EQ(0, queue.GetNumThreadsCompletingTasks());
 223     EXPECT_EQ(0, queue.task_count());
 224     EXPECT_EQ(0, queue.GetMaxCompletionsByWorkerThread());
 225     EXPECT_EQ(0, queue.GetMinCompletionsByWorkerThread());
 226     EXPECT_EQ(0, queue.GetNumberOfCompletedTasks());
 227
 228     // Set up to make each task include getting help from another worker, so
 229     // so that the work gets done in paralell.
 230     queue.ResetHistory();
 231     queue.SetTaskCount(kTaskCount);
 232     queue.SetWorkTime(kThirtyMs);
 233     queue.SetAllowHelp(true);
 234
 235     start_time = Time::Now();
 236   }
 237
 238   queue.work_is_available()->Signal();  // But each worker can signal another.
 239   // Wait till we at least start to handle tasks (and we're not all waiting).
 240   queue.SpinUntilTaskCountLessThan(kTaskCount);
 241   // Wait to allow the all workers to get done.
 242   queue.SpinUntilAllThreadsAreWaiting();
 243
 244   {
 245     // Wait until all work tasks have at least been assigned.
 246     base::AutoLock auto_lock(*queue.lock());
 247     while (queue.task_count())
 248       queue.no_more_tasks()->Wait();
 249
 250     // To avoid racy assumptions, we'll just assert that at least 2 threads
 251     // did work.  We know that the first worker should have gone to sleep, and
 252     // hence a second worker should have gotten an assignment.
 253     EXPECT_LE(2, queue.GetNumThreadsTakingAssignments());
 254     EXPECT_EQ(kTaskCount, queue.GetNumberOfCompletedTasks());
 255
 256     // Try to ask all workers to help, and only a few will do the work.
 257     queue.ResetHistory();
 258     queue.SetTaskCount(3);
 259     queue.SetWorkTime(kThirtyMs);
 260     queue.SetAllowHelp(false);
 261   }
 262   queue.work_is_available()->Broadcast();  // Make them all try.
 263   // Wait till we at least start to handle tasks (and we're not all waiting).
 264   queue.SpinUntilTaskCountLessThan(3);
 265   // Wait to allow the 3 workers to get done.
 266   queue.SpinUntilAllThreadsAreWaiting();
 267
 268   {
 269     base::AutoLock auto_lock(*queue.lock());
 270     EXPECT_EQ(3, queue.GetNumThreadsTakingAssignments());
 271     EXPECT_EQ(3, queue.GetNumThreadsCompletingTasks());
 272     EXPECT_EQ(0, queue.task_count());
 273     EXPECT_EQ(1, queue.GetMaxCompletionsByWorkerThread());
 274     EXPECT_EQ(0, queue.GetMinCompletionsByWorkerThread());
 275     EXPECT_EQ(3, queue.GetNumberOfCompletedTasks());
 276
 277     // Set up to make each task get help from another worker.
 278     queue.ResetHistory();
 279     queue.SetTaskCount(3);
 280     queue.SetWorkTime(kThirtyMs);
 281     queue.SetAllowHelp(true);  // Allow (unnecessary) help requests.
 282   }
 283   queue.work_is_available()->Broadcast();  // Signal all threads.
 284   // Wait till we at least start to handle tasks (and we're not all waiting).
 285   queue.SpinUntilTaskCountLessThan(3);
 286   // Wait to allow the 3 workers to get done.
 287   queue.SpinUntilAllThreadsAreWaiting();
 288
 289   {
 290     base::AutoLock auto_lock(*queue.lock());
 291     EXPECT_EQ(3, queue.GetNumThreadsTakingAssignments());
 292     EXPECT_EQ(3, queue.GetNumThreadsCompletingTasks());
 293     EXPECT_EQ(0, queue.task_count());
 294     EXPECT_EQ(1, queue.GetMaxCompletionsByWorkerThread());
 295     EXPECT_EQ(0, queue.GetMinCompletionsByWorkerThread());
 296     EXPECT_EQ(3, queue.GetNumberOfCompletedTasks());
 297
 298     // Set up to make each task get help from another worker.
 299     queue.ResetHistory();
 300     queue.SetTaskCount(20);  // 2 tasks per thread.
 301     queue.SetWorkTime(kThirtyMs);
 302     queue.SetAllowHelp(true);
 303   }
 304   queue.work_is_available()->Signal();  // But each worker can signal another.
 305   // Wait till we at least start to handle tasks (and we're not all waiting).
 306   queue.SpinUntilTaskCountLessThan(20);
 307   // Wait to allow the 10 workers to get done.
 308   queue.SpinUntilAllThreadsAreWaiting();  // Should take about 60 ms.
 309
 310   {
 311     base::AutoLock auto_lock(*queue.lock());
 312     EXPECT_EQ(10, queue.GetNumThreadsTakingAssignments());
 313     EXPECT_EQ(10, queue.GetNumThreadsCompletingTasks());
 314     EXPECT_EQ(0, queue.task_count());
 315     EXPECT_EQ(20, queue.GetNumberOfCompletedTasks());
 316
 317     // Same as last test, but with Broadcast().
 318     queue.ResetHistory();
 319     queue.SetTaskCount(20);  // 2 tasks per thread.
 320     queue.SetWorkTime(kThirtyMs);
 321     queue.SetAllowHelp(true);
 322   }
 323   queue.work_is_available()->Broadcast();
 324   // Wait till we at least start to handle tasks (and we're not all waiting).
 325   queue.SpinUntilTaskCountLessThan(20);
 326   // Wait to allow the 10 workers to get done.
 327   queue.SpinUntilAllThreadsAreWaiting();  // Should take about 60 ms.
 328
 329   {
 330     base::AutoLock auto_lock(*queue.lock());
 331     EXPECT_EQ(10, queue.GetNumThreadsTakingAssignments());
 332     EXPECT_EQ(10, queue.GetNumThreadsCompletingTasks());
 333     EXPECT_EQ(0, queue.task_count());
 334     EXPECT_EQ(20, queue.GetNumberOfCompletedTasks());
 335
 336     queue.SetShutdown();
 337   }
 338   queue.work_is_available()->Broadcast();  // Force check for shutdown.
 339
 340   SPIN_FOR_TIMEDELTA_OR_UNTIL_TRUE(TimeDelta::FromMinutes(1),
 341                                    queue.ThreadSafeCheckShutdown(kThreadCount));
 342 }
 343
 344 TEST_F(ConditionVariableTest, LargeFastTaskTest) {
 345   const int kThreadCount = 200;
 346   WorkQueue queue(kThreadCount);  // Start the threads.
 347
 348   Lock private_lock;  // Used locally for master to wait.
 349   base::AutoLock private_held_lock(private_lock);
 350   ConditionVariable private_cv(&private_lock);
 351
 352   {
 353     base::AutoLock auto_lock(*queue.lock());
 354     while (!queue.EveryIdWasAllocated())
 355       queue.all_threads_have_ids()->Wait();
 356   }
 357
 358   // Wait a bit more to allow threads to reach their wait state.
 359   queue.SpinUntilAllThreadsAreWaiting();
 360
 361   {
 362     // Since we have no tasks, all threads should be waiting by now.
 363     base::AutoLock auto_lock(*queue.lock());
 364     EXPECT_EQ(0, queue.GetNumThreadsTakingAssignments());
 365     EXPECT_EQ(0, queue.GetNumThreadsCompletingTasks());
 366     EXPECT_EQ(0, queue.task_count());
 367     EXPECT_EQ(0, queue.GetMaxCompletionsByWorkerThread());
 368     EXPECT_EQ(0, queue.GetMinCompletionsByWorkerThread());
 369     EXPECT_EQ(0, queue.GetNumberOfCompletedTasks());
 370
 371     // Set up to make all workers do (an average of) 20 tasks.
 372     queue.ResetHistory();
 373     queue.SetTaskCount(20 * kThreadCount);
 374     queue.SetWorkTime(kFortyFiveMs);
 375     queue.SetAllowHelp(false);
 376   }
 377   queue.work_is_available()->Broadcast();  // Start up all threads.
 378   // Wait until we've handed out all tasks.
 379   {
 380     base::AutoLock auto_lock(*queue.lock());
 381     while (queue.task_count() != 0)
 382       queue.no_more_tasks()->Wait();
 383   }
 384
 385   // Wait till the last of the tasks complete.
 386   queue.SpinUntilAllThreadsAreWaiting();
 387
 388   {
 389     // With Broadcast(), every thread should have participated.
 390     // but with racing.. they may not all have done equal numbers of tasks.
 391     base::AutoLock auto_lock(*queue.lock());
 392     EXPECT_EQ(kThreadCount, queue.GetNumThreadsTakingAssignments());
 393     EXPECT_EQ(kThreadCount, queue.GetNumThreadsCompletingTasks());
 394     EXPECT_EQ(0, queue.task_count());
 395     EXPECT_LE(20, queue.GetMaxCompletionsByWorkerThread());
 396     EXPECT_EQ(20 * kThreadCount, queue.GetNumberOfCompletedTasks());
 397
 398     // Set up to make all workers do (an average of) 4 tasks.
 399     queue.ResetHistory();
 400     queue.SetTaskCount(kThreadCount * 4);
 401     queue.SetWorkTime(kFortyFiveMs);
 402     queue.SetAllowHelp(true);  // Might outperform Broadcast().
 403   }
 404   queue.work_is_available()->Signal();  // Start up one thread.
 405
 406   // Wait until we've handed out all tasks
 407   {
 408     base::AutoLock auto_lock(*queue.lock());
 409     while (queue.task_count() != 0)
 410       queue.no_more_tasks()->Wait();
 411   }
 412
 413   // Wait till the last of the tasks complete.
 414   queue.SpinUntilAllThreadsAreWaiting();
 415
 416   {
 417     // With Signal(), every thread should have participated.
 418     // but with racing.. they may not all have done four tasks.
 419     base::AutoLock auto_lock(*queue.lock());
 420     EXPECT_EQ(kThreadCount, queue.GetNumThreadsTakingAssignments());
 421     EXPECT_EQ(kThreadCount, queue.GetNumThreadsCompletingTasks());
 422     EXPECT_EQ(0, queue.task_count());
 423     EXPECT_LE(4, queue.GetMaxCompletionsByWorkerThread());
 424     EXPECT_EQ(4 * kThreadCount, queue.GetNumberOfCompletedTasks());
 425
 426     queue.SetShutdown();
 427   }
 428   queue.work_is_available()->Broadcast();  // Force check for shutdown.
 429
 430   // Wait for shutdowns to complete.
 431   SPIN_FOR_TIMEDELTA_OR_UNTIL_TRUE(TimeDelta::FromMinutes(1),
 432                                    queue.ThreadSafeCheckShutdown(kThreadCount));
 433 }
 434
 435 //------------------------------------------------------------------------------
 436 // Finally we provide the implementation for the methods in the WorkQueue class.
 437 //------------------------------------------------------------------------------
 438
 439 WorkQueue::WorkQueue(int thread_count)
 440   : lock_(),
 441     work_is_available_(&lock_),
 442     all_threads_have_ids_(&lock_),
 443     no_more_tasks_(&lock_),
 444     thread_count_(thread_count),
 445     waiting_thread_count_(0),
 446     thread_handles_(new PlatformThreadHandle[thread_count]),
 447     assignment_history_(thread_count),
 448     completion_history_(thread_count),
 449     thread_started_counter_(0),
 450     shutdown_task_count_(0),
 451     task_count_(0),
 452     allow_help_requests_(false),
 453     shutdown_(false) {
 454   EXPECT_GE(thread_count_, 1);
 455   ResetHistory();
 456   SetTaskCount(0);
 457   SetWorkTime(TimeDelta::FromMilliseconds(30));
 458
 459   for (int i = 0; i < thread_count_; ++i) {
 460     PlatformThreadHandle pth;
 461     EXPECT_TRUE(PlatformThread::Create(0, this, &pth));
 462     thread_handles_[i] = pth;
 463   }
 464 }
 465
 466 WorkQueue::~WorkQueue() {
 467   {
 468     base::AutoLock auto_lock(lock_);
 469     SetShutdown();
 470   }
 471   work_is_available_.Broadcast();  // Tell them all to terminate.
 472
 473   for (int i = 0; i < thread_count_; ++i) {
 474     PlatformThread::Join(thread_handles_[i]);
 475   }
 476   EXPECT_EQ(0, waiting_thread_count_);
 477 }
 478
 479 int WorkQueue::GetThreadId() {
 480   DFAKE_SCOPED_RECURSIVE_LOCK(locked_methods_);
 481   DCHECK(!EveryIdWasAllocated());
 482   return thread_started_counter_++;  // Give out Unique IDs.
 483 }
 484
 485 bool WorkQueue::EveryIdWasAllocated() const {
 486   DFAKE_SCOPED_RECURSIVE_LOCK(locked_methods_);
 487   return thread_count_ == thread_started_counter_;
 488 }
 489
 490 TimeDelta WorkQueue::GetAnAssignment(int thread_id) {
 491   DFAKE_SCOPED_RECURSIVE_LOCK(locked_methods_);
 492   DCHECK_LT(0, task_count_);
 493   assignment_history_[thread_id]++;
 494   if (0 == --task_count_) {
 495     no_more_tasks_.Signal();
 496   }
 497   return worker_delay_;
 498 }
 499
 500 void WorkQueue::WorkIsCompleted(int thread_id) {
 501   DFAKE_SCOPED_RECURSIVE_LOCK(locked_methods_);
 502   completion_history_[thread_id]++;
 503 }
 504
 505 int WorkQueue::task_count() const {
 506   DFAKE_SCOPED_RECURSIVE_LOCK(locked_methods_);
 507   return task_count_;
 508 }
 509
 510 bool WorkQueue::allow_help_requests() const {
 511   DFAKE_SCOPED_RECURSIVE_LOCK(locked_methods_);
 512   return allow_help_requests_;
 513 }
 514
 515 bool WorkQueue::shutdown() const {
 516   lock_.AssertAcquired();
 517   DFAKE_SCOPED_RECURSIVE_LOCK(locked_methods_);
 518   return shutdown_;
 519 }
 520
 521 // Because this method is called from the test's main thread we need to actually
 522 // take the lock.  Threads will call the thread_shutting_down() method with the
 523 // lock already acquired.
 524 bool WorkQueue::ThreadSafeCheckShutdown(int thread_count) {
 525   bool all_shutdown;
 526   base::AutoLock auto_lock(lock_);
 527   {
 528     // Declare in scope so DFAKE is guranteed to be destroyed before AutoLock.
 529     DFAKE_SCOPED_RECURSIVE_LOCK(locked_methods_);
 530     all_shutdown = (shutdown_task_count_ == thread_count);
 531   }
 532   return all_shutdown;
 533 }
 534
 535 void WorkQueue::thread_shutting_down() {
 536   lock_.AssertAcquired();
 537   DFAKE_SCOPED_RECURSIVE_LOCK(locked_methods_);
 538   shutdown_task_count_++;
 539 }
 540
 541 Lock* WorkQueue::lock() {
 542   return &lock_;
 543 }
 544
 545 ConditionVariable* WorkQueue::work_is_available() {
 546   return &work_is_available_;
 547 }
 548
 549 ConditionVariable* WorkQueue::all_threads_have_ids() {
 550   return &all_threads_have_ids_;
 551 }
 552
 553 ConditionVariable* WorkQueue::no_more_tasks() {
 554   return &no_more_tasks_;
 555 }
 556
 557 void WorkQueue::ResetHistory() {
 558   for (int i = 0; i < thread_count_; ++i) {
 559     assignment_history_[i] = 0;
 560     completion_history_[i] = 0;
 561   }
 562 }
 563
 564 int WorkQueue::GetMinCompletionsByWorkerThread() const {
 565   int minumum = completion_history_[0];
 566   for (int i = 0; i < thread_count_; ++i)
 567     minumum = std::min(minumum, completion_history_[i]);
 568   return minumum;
 569 }
 570
 571 int WorkQueue::GetMaxCompletionsByWorkerThread() const {
 572   int maximum = completion_history_[0];
 573   for (int i = 0; i < thread_count_; ++i)
 574     maximum = std::max(maximum, completion_history_[i]);
 575   return maximum;
 576 }
 577
 578 int WorkQueue::GetNumThreadsTakingAssignments() const {
 579   int count = 0;
 580   for (int i = 0; i < thread_count_; ++i)
 581     if (assignment_history_[i])
 582       count++;
 583   return count;
 584 }
 585
 586 int WorkQueue::GetNumThreadsCompletingTasks() const {
 587   int count = 0;
 588   for (int i = 0; i < thread_count_; ++i)
 589     if (completion_history_[i])
 590       count++;
 591   return count;
 592 }
 593
 594 int WorkQueue::GetNumberOfCompletedTasks() const {
 595   int total = 0;
 596   for (int i = 0; i < thread_count_; ++i)
 597     total += completion_history_[i];
 598   return total;
 599 }
 600
 601 TimeDelta WorkQueue::GetWorkTime() const {
 602   return worker_delay_;
 603 }
 604
 605 void WorkQueue::SetWorkTime(TimeDelta delay) {
 606   worker_delay_ = delay;
 607 }
 608
 609 void WorkQueue::SetTaskCount(int count) {
 610   task_count_ = count;
 611 }
 612
 613 void WorkQueue::SetAllowHelp(bool allow) {
 614   allow_help_requests_ = allow;
 615 }
 616
 617 void WorkQueue::SetShutdown() {
 618   lock_.AssertAcquired();
 619   shutdown_ = true;
 620 }
 621
 622 void WorkQueue::SpinUntilAllThreadsAreWaiting() {
 623   while (true) {
 624     {
 625       base::AutoLock auto_lock(lock_);
 626       if (waiting_thread_count_ == thread_count_)
 627         break;
 628     }
 629     PlatformThread::Sleep(TimeDelta::FromMilliseconds(30));
 630   }
 631 }
 632
 633 void WorkQueue::SpinUntilTaskCountLessThan(int task_count) {
 634   while (true) {
 635     {
 636       base::AutoLock auto_lock(lock_);
 637       if (task_count_ < task_count)
 638         break;
 639     }
 640     PlatformThread::Sleep(TimeDelta::FromMilliseconds(30));
 641   }
 642 }
 643
 644
 645 //------------------------------------------------------------------------------
 646 // Define the standard worker task. Several tests will spin out many of these
 647 // threads.
 648 //------------------------------------------------------------------------------
 649
 650 // The multithread tests involve several threads with a task to perform as
 651 // directed by an instance of the class WorkQueue.
 652 // The task is to:
 653 // a) Check to see if there are more tasks (there is a task counter).
 654 //    a1) Wait on condition variable if there are no tasks currently.
 655 // b) Call a function to see what should be done.
 656 // c) Do some computation based on the number of milliseconds returned in (b).
 657 // d) go back to (a).
 658
 659 // WorkQueue::ThreadMain() implements the above task for all threads.
 660 // It calls the controlling object to tell the creator about progress, and to
 661 // ask about tasks.
 662
 663 void WorkQueue::ThreadMain() {
 664   int thread_id;
 665   {
 666     base::AutoLock auto_lock(lock_);
 667     thread_id = GetThreadId();
 668     if (EveryIdWasAllocated())
 669       all_threads_have_ids()->Signal();  // Tell creator we're ready.
 670   }
 671
 672   Lock private_lock;  // Used to waste time on "our work".
 673   while (1) {  // This is the main consumer loop.
 674     TimeDelta work_time;
 675     bool could_use_help;
 676     {
 677       base::AutoLock auto_lock(lock_);
 678       while (0 == task_count() && !shutdown()) {
 679         ++waiting_thread_count_;
 680         work_is_available()->Wait();
 681         --waiting_thread_count_;
 682       }
 683       if (shutdown()) {
 684         // Ack the notification of a shutdown message back to the controller.
 685         thread_shutting_down();
 686         return;  // Terminate.
 687       }
 688       // Get our task duration from the queue.
 689       work_time = GetAnAssignment(thread_id);
 690       could_use_help = (task_count() > 0) && allow_help_requests();
 691     }  // Release lock
 692
 693     // Do work (outside of locked region.
 694     if (could_use_help)
 695       work_is_available()->Signal();  // Get help from other threads.
 696
 697     if (work_time > TimeDelta::FromMilliseconds(0)) {
 698       // We could just sleep(), but we'll instead further exercise the
 699       // condition variable class, and do a timed wait.
 700       base::AutoLock auto_lock(private_lock);
 701       ConditionVariable private_cv(&private_lock);
 702       private_cv.TimedWait(work_time);  // Unsynchronized waiting.
 703     }
 704
 705     {
 706       base::AutoLock auto_lock(lock_);
 707       // Send notification that we completed our "work."
 708       WorkIsCompleted(thread_id);
 709     }
 710   }
 711 }
 712
 713 }  // namespace
 714
 715 }  // namespace base