base/synchronization/condition_variable_unittest.cc - chromium/src - Git at Google

 // Copyright (c) 2012 The Chromium Authors. All rights reserved.
 // Use of this source code is governed by a BSD-style license that can be
 // found in the LICENSE file.

 // Multi-threaded tests of ConditionVariable class.

 #include "base/synchronization/condition_variable.h"

 #include <time.h>

 #include <algorithm>
 #include <memory>
 #include <vector>

 #include "base/bind.h"
 #include "base/location.h"
 #include "base/logging.h"
 #include "base/single_thread_task_runner.h"
 #include "base/synchronization/lock.h"
 #include "base/synchronization/spin_wait.h"
 #include "base/threading/platform_thread.h"
 #include "base/threading/thread.h"
 #include "base/threading/thread_collision_warner.h"
 #include "base/time/time.h"
 #include "build/build_config.h"
 #include "testing/gtest/include/gtest/gtest.h"
 #include "testing/platform_test.h"

 namespace base {

 namespace {
 //------------------------------------------------------------------------------
 // Define our test class, with several common variables.
 //------------------------------------------------------------------------------

 class ConditionVariableTest : public PlatformTest {
  public:
   const TimeDelta kZeroMs;
   const TimeDelta kTenMs;
   const TimeDelta kThirtyMs;
   const TimeDelta kFortyFiveMs;
   const TimeDelta kSixtyMs;
   const TimeDelta kOneHundredMs;

   ConditionVariableTest()
       : kZeroMs(TimeDelta::FromMilliseconds(0)),
         kTenMs(TimeDelta::FromMilliseconds(10)),
         kThirtyMs(TimeDelta::FromMilliseconds(30)),
         kFortyFiveMs(TimeDelta::FromMilliseconds(45)),
         kSixtyMs(TimeDelta::FromMilliseconds(60)),
         kOneHundredMs(TimeDelta::FromMilliseconds(100)) {
   }
 };

 //------------------------------------------------------------------------------
 // Define a class that will control activities an several multi-threaded tests.
 // The general structure of multi-threaded tests is that a test case will
 // construct an instance of a WorkQueue.  The WorkQueue will spin up some
 // threads and control them throughout their lifetime, as well as maintaining
 // a central repository of the work thread's activity.  Finally, the WorkQueue
 // will command the the worker threads to terminate.  At that point, the test
 // cases will validate that the WorkQueue has records showing that the desired
 // activities were performed.
 //------------------------------------------------------------------------------

 // Callers are responsible for synchronizing access to the following class.
 // The WorkQueue::lock_, as accessed via WorkQueue::lock(), should be used for
 // all synchronized access.
 class WorkQueue : public PlatformThread::Delegate {
  public:
   explicit WorkQueue(int thread_count);
   ~WorkQueue() override;

   // PlatformThread::Delegate interface.
   void ThreadMain() override;

   //----------------------------------------------------------------------------
   // Worker threads only call the following methods.
   // They should use the lock to get exclusive access.
   int GetThreadId();  // Get an ID assigned to a thread..
   bool EveryIdWasAllocated() const;  // Indicates that all IDs were handed out.
   TimeDelta GetAnAssignment(int thread_id);  // Get a work task duration.
   void WorkIsCompleted(int thread_id);

   int task_count() const;
   bool allow_help_requests() const;  // Workers can signal more workers.
   bool shutdown() const;  // Check if shutdown has been requested.

   void thread_shutting_down();


   //----------------------------------------------------------------------------
   // Worker threads can call them but not needed to acquire a lock.
   Lock* lock();

   ConditionVariable* work_is_available();
   ConditionVariable* all_threads_have_ids();
   ConditionVariable* no_more_tasks();

   //----------------------------------------------------------------------------
   // The rest of the methods are for use by the controlling master thread (the
   // test case code).
   void ResetHistory();
   int GetMinCompletionsByWorkerThread() const;
   int GetMaxCompletionsByWorkerThread() const;
   int GetNumThreadsTakingAssignments() const;
   int GetNumThreadsCompletingTasks() const;
   int GetNumberOfCompletedTasks() const;

   void SetWorkTime(TimeDelta delay);
   void SetTaskCount(int count);
   void SetAllowHelp(bool allow);

   // The following must be called without locking, and will spin wait until the
   // threads are all in a wait state.
   void SpinUntilAllThreadsAreWaiting();
   void SpinUntilTaskCountLessThan(int task_count);

   // Caller must acquire lock before calling.
   void SetShutdown();

   // Compares the |shutdown_task_count_| to the |thread_count| and returns true
   // if they are equal.  This check will acquire the |lock_| so the caller
   // should not hold the lock when calling this method.
   bool ThreadSafeCheckShutdown(int thread_count);

  private:
   // Both worker threads and controller use the following to synchronize.
   Lock lock_;
   ConditionVariable work_is_available_;  // To tell threads there is work.

   // Conditions to notify the controlling process (if it is interested).
   ConditionVariable all_threads_have_ids_;  // All threads are running.
   ConditionVariable no_more_tasks_;  // Task count is zero.

   const int thread_count_;
   int waiting_thread_count_;
   std::unique_ptr<PlatformThreadHandle[]> thread_handles_;
   std::vector<int> assignment_history_;  // Number of assignment per worker.
   std::vector<int> completion_history_;  // Number of completions per worker.
   int thread_started_counter_;  // Used to issue unique id to workers.
   int shutdown_task_count_;  // Number of tasks told to shutdown
   int task_count_;  // Number of assignment tasks waiting to be processed.
   TimeDelta worker_delay_;  // Time each task takes to complete.
   bool allow_help_requests_;  // Workers can signal more workers.
   bool shutdown_;  // Set when threads need to terminate.

   DFAKE_MUTEX(locked_methods_);
 };

 //------------------------------------------------------------------------------
 // The next section contains the actual tests.
 //------------------------------------------------------------------------------

 TEST_F(ConditionVariableTest, StartupShutdownTest) {
   Lock lock;

   // First try trivial startup/shutdown.
   {
     ConditionVariable cv1(&lock);
   }  // Call for cv1 destruction.

   // Exercise with at least a few waits.
   ConditionVariable cv(&lock);

   lock.Acquire();
   cv.TimedWait(kTenMs);  // Wait for 10 ms.
   cv.TimedWait(kTenMs);  // Wait for 10 ms.
   lock.Release();

   lock.Acquire();
   cv.TimedWait(kTenMs);  // Wait for 10 ms.
   cv.TimedWait(kTenMs);  // Wait for 10 ms.
   cv.TimedWait(kTenMs);  // Wait for 10 ms.
   lock.Release();
 }  // Call for cv destruction.

 TEST_F(ConditionVariableTest, TimeoutTest) {
   Lock lock;
   ConditionVariable cv(&lock);
   lock.Acquire();

   TimeTicks start = TimeTicks::Now();
   const TimeDelta WAIT_TIME = TimeDelta::FromMilliseconds(300);
   // Allow for clocking rate granularity.
   const TimeDelta FUDGE_TIME = TimeDelta::FromMilliseconds(50);

   cv.TimedWait(WAIT_TIME + FUDGE_TIME);
   TimeDelta duration = TimeTicks::Now() - start;
   // We can't use EXPECT_GE here as the TimeDelta class does not support the
   // required stream conversion.
   EXPECT_TRUE(duration >= WAIT_TIME);

   lock.Release();
 }

 #if defined(OS_POSIX)
 const int kDiscontinuitySeconds = 2;

 void BackInTime(Lock* lock) {
   AutoLock auto_lock(*lock);

   timeval tv;
   gettimeofday(&tv, NULL);
   tv.tv_sec -= kDiscontinuitySeconds;
   settimeofday(&tv, NULL);
 }

 // Tests that TimedWait ignores changes to the system clock.
 // Test is disabled by default, because it needs to run as root to muck with the
 // system clock.
 // http://crbug.com/293736
 TEST_F(ConditionVariableTest, DISABLED_TimeoutAcrossSetTimeOfDay) {
   timeval tv;
   gettimeofday(&tv, NULL);
   tv.tv_sec += kDiscontinuitySeconds;
   if (settimeofday(&tv, NULL) < 0) {
     PLOG(ERROR) << "Could not set time of day. Run as root?";
     return;
   }

   Lock lock;
   ConditionVariable cv(&lock);
   lock.Acquire();

   Thread thread("Helper");
   thread.Start();
   thread.task_runner()->PostTask(FROM_HERE, base::BindOnce(&BackInTime, &lock));

   TimeTicks start = TimeTicks::Now();
   const TimeDelta kWaitTime = TimeDelta::FromMilliseconds(300);
   // Allow for clocking rate granularity.
   const TimeDelta kFudgeTime = TimeDelta::FromMilliseconds(50);

   cv.TimedWait(kWaitTime + kFudgeTime);
   TimeDelta duration = TimeTicks::Now() - start;

   thread.Stop();
   // We can't use EXPECT_GE here as the TimeDelta class does not support the
   // required stream conversion.
   EXPECT_TRUE(duration >= kWaitTime);
   EXPECT_TRUE(duration <= TimeDelta::FromSeconds(kDiscontinuitySeconds));

   lock.Release();
 }
 #endif


 // Suddenly got flaky on Win, see http://crbug.com/10607 (starting at
 // comment #15).
 #if defined(OS_WIN)
 #define MAYBE_MultiThreadConsumerTest DISABLED_MultiThreadConsumerTest
 #else
 #define MAYBE_MultiThreadConsumerTest MultiThreadConsumerTest
 #endif
 // Test serial task servicing, as well as two parallel task servicing methods.
 TEST_F(ConditionVariableTest, MAYBE_MultiThreadConsumerTest) {
   const int kThreadCount = 10;
   WorkQueue queue(kThreadCount);  // Start the threads.

   const int kTaskCount = 10;  // Number of tasks in each mini-test here.

   Time start_time;  // Used to time task processing.

   {
     base::AutoLock auto_lock(*queue.lock());
     while (!queue.EveryIdWasAllocated())
       queue.all_threads_have_ids()->Wait();
   }

   // If threads aren't in a wait state, they may start to gobble up tasks in
   // parallel, short-circuiting (breaking) this test.
   queue.SpinUntilAllThreadsAreWaiting();

   {
     // Since we have no tasks yet, all threads should be waiting by now.
     base::AutoLock auto_lock(*queue.lock());
     EXPECT_EQ(0, queue.GetNumThreadsTakingAssignments());
     EXPECT_EQ(0, queue.GetNumThreadsCompletingTasks());
     EXPECT_EQ(0, queue.task_count());
     EXPECT_EQ(0, queue.GetMaxCompletionsByWorkerThread());
     EXPECT_EQ(0, queue.GetMinCompletionsByWorkerThread());
     EXPECT_EQ(0, queue.GetNumberOfCompletedTasks());

     // Set up to make each task include getting help from another worker, so
     // so that the work gets done in paralell.
     queue.ResetHistory();
     queue.SetTaskCount(kTaskCount);
     queue.SetWorkTime(kThirtyMs);
     queue.SetAllowHelp(true);

     start_time = Time::Now();
   }

   queue.work_is_available()->Signal();  // But each worker can signal another.
   // Wait till we at least start to handle tasks (and we're not all waiting).
   queue.SpinUntilTaskCountLessThan(kTaskCount);
   // Wait to allow the all workers to get done.
   queue.SpinUntilAllThreadsAreWaiting();

   {
     // Wait until all work tasks have at least been assigned.
     base::AutoLock auto_lock(*queue.lock());
     while (queue.task_count())
       queue.no_more_tasks()->Wait();

     // To avoid racy assumptions, we'll just assert that at least 2 threads
     // did work.  We know that the first worker should have gone to sleep, and
     // hence a second worker should have gotten an assignment.
     EXPECT_LE(2, queue.GetNumThreadsTakingAssignments());
     EXPECT_EQ(kTaskCount, queue.GetNumberOfCompletedTasks());

     // Try to ask all workers to help, and only a few will do the work.
     queue.ResetHistory();
     queue.SetTaskCount(3);
     queue.SetWorkTime(kThirtyMs);
     queue.SetAllowHelp(false);
   }
   queue.work_is_available()->Broadcast();  // Make them all try.
   // Wait till we at least start to handle tasks (and we're not all waiting).
   queue.SpinUntilTaskCountLessThan(3);
   // Wait to allow the 3 workers to get done.
   queue.SpinUntilAllThreadsAreWaiting();

   {
     base::AutoLock auto_lock(*queue.lock());
     EXPECT_EQ(3, queue.GetNumThreadsTakingAssignments());
     EXPECT_EQ(3, queue.GetNumThreadsCompletingTasks());
     EXPECT_EQ(0, queue.task_count());
     EXPECT_EQ(1, queue.GetMaxCompletionsByWorkerThread());
     EXPECT_EQ(0, queue.GetMinCompletionsByWorkerThread());
     EXPECT_EQ(3, queue.GetNumberOfCompletedTasks());

     // Set up to make each task get help from another worker.
     queue.ResetHistory();
     queue.SetTaskCount(3);
     queue.SetWorkTime(kThirtyMs);
     queue.SetAllowHelp(true);  // Allow (unnecessary) help requests.
   }
   queue.work_is_available()->Broadcast();  // Signal all threads.
   // Wait till we at least start to handle tasks (and we're not all waiting).
   queue.SpinUntilTaskCountLessThan(3);
   // Wait to allow the 3 workers to get done.
   queue.SpinUntilAllThreadsAreWaiting();

   {
     base::AutoLock auto_lock(*queue.lock());
     EXPECT_EQ(3, queue.GetNumThreadsTakingAssignments());
     EXPECT_EQ(3, queue.GetNumThreadsCompletingTasks());
     EXPECT_EQ(0, queue.task_count());
     EXPECT_EQ(1, queue.GetMaxCompletionsByWorkerThread());
     EXPECT_EQ(0, queue.GetMinCompletionsByWorkerThread());
     EXPECT_EQ(3, queue.GetNumberOfCompletedTasks());

     // Set up to make each task get help from another worker.
     queue.ResetHistory();
     queue.SetTaskCount(20);  // 2 tasks per thread.
     queue.SetWorkTime(kThirtyMs);
     queue.SetAllowHelp(true);
   }
   queue.work_is_available()->Signal();  // But each worker can signal another.
   // Wait till we at least start to handle tasks (and we're not all waiting).
   queue.SpinUntilTaskCountLessThan(20);
   // Wait to allow the 10 workers to get done.
   queue.SpinUntilAllThreadsAreWaiting();  // Should take about 60 ms.

   {
     base::AutoLock auto_lock(*queue.lock());
     EXPECT_EQ(10, queue.GetNumThreadsTakingAssignments());
     EXPECT_EQ(10, queue.GetNumThreadsCompletingTasks());
     EXPECT_EQ(0, queue.task_count());
     EXPECT_EQ(20, queue.GetNumberOfCompletedTasks());

     // Same as last test, but with Broadcast().
     queue.ResetHistory();
     queue.SetTaskCount(20);  // 2 tasks per thread.
     queue.SetWorkTime(kThirtyMs);
     queue.SetAllowHelp(true);
   }
   queue.work_is_available()->Broadcast();
   // Wait till we at least start to handle tasks (and we're not all waiting).
   queue.SpinUntilTaskCountLessThan(20);
   // Wait to allow the 10 workers to get done.
   queue.SpinUntilAllThreadsAreWaiting();  // Should take about 60 ms.

   {
     base::AutoLock auto_lock(*queue.lock());
     EXPECT_EQ(10, queue.GetNumThreadsTakingAssignments());
     EXPECT_EQ(10, queue.GetNumThreadsCompletingTasks());
     EXPECT_EQ(0, queue.task_count());
     EXPECT_EQ(20, queue.GetNumberOfCompletedTasks());

     queue.SetShutdown();
   }
   queue.work_is_available()->Broadcast();  // Force check for shutdown.

   SPIN_FOR_TIMEDELTA_OR_UNTIL_TRUE(TimeDelta::FromMinutes(1),
                                    queue.ThreadSafeCheckShutdown(kThreadCount));
 }

 TEST_F(ConditionVariableTest, LargeFastTaskTest) {
   const int kThreadCount = 200;
   WorkQueue queue(kThreadCount);  // Start the threads.

   Lock private_lock;  // Used locally for master to wait.
   base::AutoLock private_held_lock(private_lock);
   ConditionVariable private_cv(&private_lock);

   {
     base::AutoLock auto_lock(*queue.lock());
     while (!queue.EveryIdWasAllocated())
       queue.all_threads_have_ids()->Wait();
   }

   // Wait a bit more to allow threads to reach their wait state.
   queue.SpinUntilAllThreadsAreWaiting();

   {
     // Since we have no tasks, all threads should be waiting by now.
     base::AutoLock auto_lock(*queue.lock());
     EXPECT_EQ(0, queue.GetNumThreadsTakingAssignments());
     EXPECT_EQ(0, queue.GetNumThreadsCompletingTasks());
     EXPECT_EQ(0, queue.task_count());
     EXPECT_EQ(0, queue.GetMaxCompletionsByWorkerThread());
     EXPECT_EQ(0, queue.GetMinCompletionsByWorkerThread());
     EXPECT_EQ(0, queue.GetNumberOfCompletedTasks());

     // Set up to make all workers do (an average of) 20 tasks.
     queue.ResetHistory();
     queue.SetTaskCount(20 * kThreadCount);
     queue.SetWorkTime(kFortyFiveMs);
     queue.SetAllowHelp(false);
   }
   queue.work_is_available()->Broadcast();  // Start up all threads.
   // Wait until we've handed out all tasks.
   {
     base::AutoLock auto_lock(*queue.lock());
     while (queue.task_count() != 0)
       queue.no_more_tasks()->Wait();
   }

   // Wait till the last of the tasks complete.
   queue.SpinUntilAllThreadsAreWaiting();

   {
     // With Broadcast(), every thread should have participated.
     // but with racing.. they may not all have done equal numbers of tasks.
     base::AutoLock auto_lock(*queue.lock());
     EXPECT_EQ(kThreadCount, queue.GetNumThreadsTakingAssignments());
     EXPECT_EQ(kThreadCount, queue.GetNumThreadsCompletingTasks());
     EXPECT_EQ(0, queue.task_count());
     EXPECT_LE(20, queue.GetMaxCompletionsByWorkerThread());
     EXPECT_EQ(20 * kThreadCount, queue.GetNumberOfCompletedTasks());

     // Set up to make all workers do (an average of) 4 tasks.
     queue.ResetHistory();
     queue.SetTaskCount(kThreadCount * 4);
     queue.SetWorkTime(kFortyFiveMs);
     queue.SetAllowHelp(true);  // Might outperform Broadcast().
   }
   queue.work_is_available()->Signal();  // Start up one thread.

   // Wait until we've handed out all tasks
   {
     base::AutoLock auto_lock(*queue.lock());
     while (queue.task_count() != 0)
       queue.no_more_tasks()->Wait();
   }

   // Wait till the last of the tasks complete.
   queue.SpinUntilAllThreadsAreWaiting();

   {
     // With Signal(), every thread should have participated.
     // but with racing.. they may not all have done four tasks.
     base::AutoLock auto_lock(*queue.lock());
     EXPECT_EQ(kThreadCount, queue.GetNumThreadsTakingAssignments());
     EXPECT_EQ(kThreadCount, queue.GetNumThreadsCompletingTasks());
     EXPECT_EQ(0, queue.task_count());
     EXPECT_LE(4, queue.GetMaxCompletionsByWorkerThread());
     EXPECT_EQ(4 * kThreadCount, queue.GetNumberOfCompletedTasks());

     queue.SetShutdown();
   }
   queue.work_is_available()->Broadcast();  // Force check for shutdown.

   // Wait for shutdowns to complete.
   SPIN_FOR_TIMEDELTA_OR_UNTIL_TRUE(TimeDelta::FromMinutes(1),
                                    queue.ThreadSafeCheckShutdown(kThreadCount));
 }

 //------------------------------------------------------------------------------
 // Finally we provide the implementation for the methods in the WorkQueue class.
 //------------------------------------------------------------------------------

 WorkQueue::WorkQueue(int thread_count)
   : lock_(),
     work_is_available_(&lock_),
     all_threads_have_ids_(&lock_),
     no_more_tasks_(&lock_),
     thread_count_(thread_count),
     waiting_thread_count_(0),
     thread_handles_(new PlatformThreadHandle[thread_count]),
     assignment_history_(thread_count),
     completion_history_(thread_count),
     thread_started_counter_(0),
     shutdown_task_count_(0),
     task_count_(0),
     allow_help_requests_(false),
     shutdown_(false) {
   EXPECT_GE(thread_count_, 1);
   ResetHistory();
   SetTaskCount(0);
   SetWorkTime(TimeDelta::FromMilliseconds(30));

   for (int i = 0; i < thread_count_; ++i) {
     PlatformThreadHandle pth;
     EXPECT_TRUE(PlatformThread::Create(0, this, &pth));
     thread_handles_[i] = pth;
   }
 }

 WorkQueue::~WorkQueue() {
   {
     base::AutoLock auto_lock(lock_);
     SetShutdown();
   }
   work_is_available_.Broadcast();  // Tell them all to terminate.

   for (int i = 0; i < thread_count_; ++i) {
     PlatformThread::Join(thread_handles_[i]);
   }
   EXPECT_EQ(0, waiting_thread_count_);
 }

 int WorkQueue::GetThreadId() {
   DFAKE_SCOPED_RECURSIVE_LOCK(locked_methods_);
   DCHECK(!EveryIdWasAllocated());
   return thread_started_counter_++;  // Give out Unique IDs.
 }

 bool WorkQueue::EveryIdWasAllocated() const {
   DFAKE_SCOPED_RECURSIVE_LOCK(locked_methods_);
   return thread_count_ == thread_started_counter_;
 }

 TimeDelta WorkQueue::GetAnAssignment(int thread_id) {
   DFAKE_SCOPED_RECURSIVE_LOCK(locked_methods_);
   DCHECK_LT(0, task_count_);
   assignment_history_[thread_id]++;
   if (0 == --task_count_) {
     no_more_tasks_.Signal();
   }
   return worker_delay_;
 }

 void WorkQueue::WorkIsCompleted(int thread_id) {
   DFAKE_SCOPED_RECURSIVE_LOCK(locked_methods_);
   completion_history_[thread_id]++;
 }

 int WorkQueue::task_count() const {
   DFAKE_SCOPED_RECURSIVE_LOCK(locked_methods_);
   return task_count_;
 }

 bool WorkQueue::allow_help_requests() const {
   DFAKE_SCOPED_RECURSIVE_LOCK(locked_methods_);
   return allow_help_requests_;
 }

 bool WorkQueue::shutdown() const {
   lock_.AssertAcquired();
   DFAKE_SCOPED_RECURSIVE_LOCK(locked_methods_);
   return shutdown_;
 }

 // Because this method is called from the test's main thread we need to actually
 // take the lock.  Threads will call the thread_shutting_down() method with the
 // lock already acquired.
 bool WorkQueue::ThreadSafeCheckShutdown(int thread_count) {
   bool all_shutdown;
   base::AutoLock auto_lock(lock_);
   {
     // Declare in scope so DFAKE is guranteed to be destroyed before AutoLock.
     DFAKE_SCOPED_RECURSIVE_LOCK(locked_methods_);
     all_shutdown = (shutdown_task_count_ == thread_count);
   }
   return all_shutdown;
 }

 void WorkQueue::thread_shutting_down() {
   lock_.AssertAcquired();
   DFAKE_SCOPED_RECURSIVE_LOCK(locked_methods_);
   shutdown_task_count_++;
 }

 Lock* WorkQueue::lock() {
   return &lock_;
 }

 ConditionVariable* WorkQueue::work_is_available() {
   return &work_is_available_;
 }

 ConditionVariable* WorkQueue::all_threads_have_ids() {
   return &all_threads_have_ids_;
 }

 ConditionVariable* WorkQueue::no_more_tasks() {
   return &no_more_tasks_;
 }

 void WorkQueue::ResetHistory() {
   for (int i = 0; i < thread_count_; ++i) {
     assignment_history_[i] = 0;
     completion_history_[i] = 0;
   }
 }

 int WorkQueue::GetMinCompletionsByWorkerThread() const {
   int minumum = completion_history_[0];
   for (int i = 0; i < thread_count_; ++i)
     minumum = std::min(minumum, completion_history_[i]);
   return minumum;
 }

 int WorkQueue::GetMaxCompletionsByWorkerThread() const {
   int maximum = completion_history_[0];
   for (int i = 0; i < thread_count_; ++i)
     maximum = std::max(maximum, completion_history_[i]);
   return maximum;
 }

 int WorkQueue::GetNumThreadsTakingAssignments() const {
   int count = 0;
   for (int i = 0; i < thread_count_; ++i)
     if (assignment_history_[i])
       count++;
   return count;
 }

 int WorkQueue::GetNumThreadsCompletingTasks() const {
   int count = 0;
   for (int i = 0; i < thread_count_; ++i)
     if (completion_history_[i])
       count++;
   return count;
 }

 int WorkQueue::GetNumberOfCompletedTasks() const {
   int total = 0;
   for (int i = 0; i < thread_count_; ++i)
     total += completion_history_[i];
   return total;
 }

 void WorkQueue::SetWorkTime(TimeDelta delay) {
   worker_delay_ = delay;
 }

 void WorkQueue::SetTaskCount(int count) {
   task_count_ = count;
 }

 void WorkQueue::SetAllowHelp(bool allow) {
   allow_help_requests_ = allow;
 }

 void WorkQueue::SetShutdown() {
   lock_.AssertAcquired();
   shutdown_ = true;
 }

 void WorkQueue::SpinUntilAllThreadsAreWaiting() {
   while (true) {
     {
       base::AutoLock auto_lock(lock_);
       if (waiting_thread_count_ == thread_count_)
         break;
     }
     PlatformThread::Sleep(TimeDelta::FromMilliseconds(30));
   }
 }

 void WorkQueue::SpinUntilTaskCountLessThan(int task_count) {
   while (true) {
     {
       base::AutoLock auto_lock(lock_);
       if (task_count_ < task_count)
         break;
     }
     PlatformThread::Sleep(TimeDelta::FromMilliseconds(30));
   }
 }


 //------------------------------------------------------------------------------
 // Define the standard worker task. Several tests will spin out many of these
 // threads.
 //------------------------------------------------------------------------------

 // The multithread tests involve several threads with a task to perform as
 // directed by an instance of the class WorkQueue.
 // The task is to:
 // a) Check to see if there are more tasks (there is a task counter).
 //    a1) Wait on condition variable if there are no tasks currently.
 // b) Call a function to see what should be done.
 // c) Do some computation based on the number of milliseconds returned in (b).
 // d) go back to (a).

 // WorkQueue::ThreadMain() implements the above task for all threads.
 // It calls the controlling object to tell the creator about progress, and to
 // ask about tasks.

 void WorkQueue::ThreadMain() {
   int thread_id;
   {
     base::AutoLock auto_lock(lock_);
     thread_id = GetThreadId();
     if (EveryIdWasAllocated())
       all_threads_have_ids()->Signal();  // Tell creator we're ready.
   }

   Lock private_lock;  // Used to waste time on "our work".
   while (1) {  // This is the main consumer loop.
     TimeDelta work_time;
     bool could_use_help;
     {
       base::AutoLock auto_lock(lock_);
       while (0 == task_count() && !shutdown()) {
         ++waiting_thread_count_;
         work_is_available()->Wait();
         --waiting_thread_count_;
       }
       if (shutdown()) {
         // Ack the notification of a shutdown message back to the controller.
         thread_shutting_down();
         return;  // Terminate.
       }
       // Get our task duration from the queue.
       work_time = GetAnAssignment(thread_id);
       could_use_help = (task_count() > 0) && allow_help_requests();
     }  // Release lock

     // Do work (outside of locked region.
     if (could_use_help)
       work_is_available()->Signal();  // Get help from other threads.

     if (work_time > TimeDelta::FromMilliseconds(0)) {
       // We could just sleep(), but we'll instead further exercise the
       // condition variable class, and do a timed wait.
       base::AutoLock auto_lock(private_lock);
       ConditionVariable private_cv(&private_lock);
       private_cv.TimedWait(work_time);  // Unsynchronized waiting.
     }

     {
       base::AutoLock auto_lock(lock_);
       // Send notification that we completed our "work."
       WorkIsCompleted(thread_id);
     }
   }
 }

 }  // namespace

 }  // namespace base