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