text: // 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/test/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, nullptr);
    tv.tv_sec -= kDiscontinuitySeconds;
    settimeofday(&tv, nullptr);
  }
  
  // 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, nullptr);
    tv.tv_sec += kDiscontinuitySeconds;
    if (settimeofday(&tv, nullptr) < 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).
  // This is also flaky on Fuchsia, see http://crbug.com/738275.
  #if defined(OS_WIN) || defined(OS_FUCHSIA)
  #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));
  }
  
  #if defined(OS_FUCHSIA)
  // TODO(crbug.com/751894): This flakily times out on Fuchsia.
  #define MAYBE_LargeFastTaskTest DISABLED_LargeFastTaskTest
  #else
  #define MAYBE_LargeFastTaskTest LargeFastTaskTest
  #endif
  TEST_F(ConditionVariableTest, MAYBE_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
  