service_runtime/nacl_sync_unittest.cc

   1 /*
   2  * Copyright 2008, Google Inc.
   3  * All rights reserved.
   4  *
   5  * Redistribution and use in source and binary forms, with or without
   6  * modification, are permitted provided that the following conditions are
   7  * met:
   8  *
   9  *     * Redistributions of source code must retain the above copyright
  10  * notice, this list of conditions and the following disclaimer.
  11  *     * Redistributions in binary form must reproduce the above
  12  * copyright notice, this list of conditions and the following disclaimer
  13  * in the documentation and/or other materials provided with the
  14  * distribution.
  15  *     * Neither the name of Google Inc. nor the names of its
  16  * contributors may be used to endorse or promote products derived from
  17  * this software without specific prior written permission.
  18  *
  19  * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
  20  * "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
  21  * LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
  22  * A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT
  23  * OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
  24  * SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
  25  * LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
  26  * DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
  27  * THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
  28  * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
  29  * OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
  30  */
  31
  32 #include "native_client/include/portability.h"
  33
  34 #include <time.h>
  35 #include <vector>
  36
  37 #if NACL_WINDOWS
  38 # include "native_client/service_runtime/win/condition_variable.h"
  39 #elif NACL_LINUX || NACL_OSX
  40 # include "native_client/service_runtime/linux/condition_variable.h"
  41 #endif
  42
  43 #include "gtest/gtest.h"
  44
  45 namespace {
  46
  47 #define SPIN_FOR_TIMEDELTA_OR_UNTIL_TRUE(tsp, cond) \
  48   for (int i =0; i < (tsp).InMilliseconds()/10; i++) \
  49     if (cond) \
  50       break; \
  51     else \
  52       Sleep(10);
  53
  54
  55 TEST(ConditionVariableTest, StartupShutdownTest) {
  56   const NaCl::TimeDelta TEN_MS = NaCl::TimeDelta::FromMilliseconds(10);
  57   NaCl::Lock lock;
  58
  59   // First try trivial startup/shutdown.
  60   NaCl::ConditionVariable* cv = new NaCl::ConditionVariable();
  61   delete cv;
  62
  63   // Exercise with at least a few waits.
  64   cv = new NaCl::ConditionVariable();
  65   {
  66     NaCl::AutoLock auto_lock(lock);
  67     cv->TimedWaitRel(lock, TEN_MS);  // Wait for 10 ms.
  68     cv->TimedWaitRel(lock, TEN_MS);  // Wait for 10 ms.
  69   }
  70   {
  71     NaCl::AutoLock auto_lock(lock);
  72     cv->TimedWaitRel(lock, TEN_MS);  // Wait for 10 ms.
  73     cv->TimedWaitRel(lock, TEN_MS);  // Wait for 10 ms.
  74     cv->TimedWaitRel(lock, TEN_MS);  // Wait for 10 ms.
  75   }
  76   delete cv;
  77 }
  78
  79 TEST(ConditionVariableTest, LockedExpressionTest) {
  80   int i = 0;
  81   NaCl::Lock lock;
  82
  83   // Old LOCKED_EXPRESSION macro caused syntax errors here.
  84   // ... yes... compiler will optimize this example.
  85   // Syntax error is what I'm after precluding.
  86   if (0)
  87     LOCKED_EXPRESSION(lock, i = 1);
  88   else
  89     LOCKED_EXPRESSION(lock, i = 2);
  90
  91   EXPECT_EQ(2, i);
  92 }
  93
  94 TEST(ConditionVariableTest, TimeoutTest) {
  95   NaCl::Lock lock;
  96   NaCl::ConditionVariable cv;
  97
  98   {
  99     NaCl::AutoLock auto_lock(lock);
 100     NaCl::TimeTicks start = NaCl::TimeTicks::Now();
 101     const NaCl::TimeDelta WAIT_TIME = NaCl::TimeDelta::FromMilliseconds(300);
 102     // Allow for clocking rate granularity.
 103     const NaCl::TimeDelta FUDGE_TIME = NaCl::TimeDelta::FromMilliseconds(50);
 104
 105     cv.TimedWaitRel(lock, WAIT_TIME + FUDGE_TIME);
 106     NaCl::TimeDelta duration = NaCl::TimeTicks::Now() - start;
 107     EXPECT_TRUE(duration >= WAIT_TIME);
 108   }
 109 }
 110
 111 // TODO - make the code below use cross-platform thread functions
 112 #if NACL_WINDOWS
 113 // Forward declare the worker_process task
 114 DWORD WINAPI worker_process(void* p);
 115
 116 // Caller is responsible for synchronizing access to the following class.
 117 // The cs_ member should be used for all synchronized access.
 118 class WorkQueue {
 119  public:
 120   explicit WorkQueue(int thread_count)
 121     : thread_count_(thread_count),
 122       shutdown_task_count_(0),
 123       assignment_history_(thread_count),
 124       completion_history_(thread_count),
 125       handles_(new HANDLE[thread_count]) {
 126         EXPECT_GE(thread_count_, 1);
 127         ResetHistory();
 128         SetTaskCount(0);
 129         SetWorkTime(NaCl::TimeDelta::FromMilliseconds(30));
 130         thread_started_counter_ = 0;
 131         allow_help_requests_ = false;
 132         shutdown_ = false;
 133         performance_ = true;
 134       for (int i = 0; i < thread_count_; i++) {
 135         handles_[i] = CreateThread(NULL,  // security.
 136                                   0,     // <64K stack size.
 137                                   worker_process,  // Static function.
 138                                   reinterpret_cast<void*>(this),
 139                                   0,      // Create running process.
 140                                   NULL);  // OS version of thread id.
 141         EXPECT_TRUE(NULL != handles_[i]);
 142       }
 143     }
 144
 145   ~WorkQueue() {
 146     {
 147       NaCl::AutoLock auto_lock(lock_);
 148       SetShutdown();
 149     }
 150     work_is_available_.Broadcast();  // Tell them all to terminate.
 151
 152     DWORD result = WaitForMultipleObjects(
 153                       thread_count_,
 154                       &handles_[0],
 155                       true,  // Wait for all
 156                       10000);  // Ten seconds max.
 157
 158     for (int i = 0; i < thread_count_; i++) {
 159       int ret_value = CloseHandle(handles_[i]);
 160       // CHECK(0 != ret_value);
 161       handles_[i] = NULL;
 162     }
 163     delete[] handles_;
 164   }
 165
 166   // Worker threads only call the following six methods.
 167   // They should use the lock to get exclusive access.
 168   int GetThreadId() {
 169     // DCHECK(!EveryIdWasAllocated());
 170     return thread_started_counter_++;  // Give out Unique IDs.
 171   }
 172
 173   bool EveryIdWasAllocated() {
 174     return thread_count_ == thread_started_counter_;
 175   }
 176
 177   NaCl::TimeDelta GetAnAssignment(int thread_id) {
 178     // DCHECK(0 < task_count_);
 179     assignment_history_[thread_id]++;
 180     if (0 == --task_count_) {
 181       no_more_tasks_.Signal();
 182     };
 183     return worker_delay_;
 184   }
 185
 186   void WorkIsCompleted(int thread_id) {
 187     completion_history_[thread_id]++;
 188   }
 189
 190   int GetRemainingTaskCount() {return task_count_;}
 191   bool GetAllowHelp() {return allow_help_requests_;}
 192   bool shutdown() {return shutdown_;}
 193   void thread_shutting_down() {shutdown_task_count_++;}
 194   int shutdown_task_count() { return shutdown_task_count_; }
 195
 196   // Both worker threads and controller use the following to synchronize.
 197   NaCl::Lock lock_;
 198   NaCl::Lock* lock() { return &lock_; }
 199   NaCl::ConditionVariable work_is_available_;  // To tell threads there is work.
 200   NaCl::ConditionVariable* work_is_available() {
 201     return &work_is_available_;
 202   }
 203
 204   // Conditions to tell the controlling process (if it is interested)
 205   NaCl::ConditionVariable all_threads_have_ids_;  // All threads are running.
 206   NaCl::ConditionVariable* all_threads_have_ids() {
 207     return &all_threads_have_ids_;
 208   }
 209   NaCl::ConditionVariable no_more_tasks_;  // Task count is zero.
 210   NaCl::ConditionVariable* no_more_tasks() { return &no_more_tasks_; }
 211
 212   //-------------------------------------------------------------------
 213   // The rest of the methods are for the controlling master
 214   // thread.
 215   void ResetHistory() {
 216     for (int i = 0; i < thread_count_; i++) {
 217       assignment_history_[i] = 0;
 218       completion_history_[i] = 0;
 219     }
 220   }
 221
 222   int GetMinCompletionsByWorkerThread() {
 223     int min = completion_history_[0];
 224     for (int i = 0; i < thread_count_; i++)
 225       if (min > completion_history_[i]) min = completion_history_[i];
 226     return min;
 227   }
 228
 229   int GetMaxCompletionsByWorkerThread() {
 230     int max = completion_history_[0];
 231     for (int i = 0; i < thread_count_; i++)
 232       if (max < completion_history_[i]) max = completion_history_[i];
 233     return max;
 234   }
 235
 236   int GetNumThreadsTakingAssignments() {
 237     int count = 0;
 238     for (int i = 0; i < thread_count_; i++)
 239       if (assignment_history_[i])
 240         count++;
 241     return count;
 242   }
 243
 244   int GetNumThreadsCompletingTasks() {
 245     int count = 0;
 246     for (int i = 0; i < thread_count_; i++)
 247       if (completion_history_[i])
 248         count++;
 249     return count;
 250   }
 251
 252   int GetNumberOfCompletedTasks() {
 253     int total = 0;
 254     for (int i = 0; i < thread_count_; i++)
 255       total += completion_history_[i];
 256     return total;
 257   }
 258
 259   void SetPerformance(bool performance) { performance_ = performance; }
 260   bool performance() { return performance_; }
 261
 262   void SetWorkTime(NaCl::TimeDelta delay) { worker_delay_ = delay; }
 263   void SetTaskCount(int count) { task_count_ = count; }
 264   void SetAllowHelp(bool allow) { allow_help_requests_ = allow; }
 265   void SetShutdown() { shutdown_ = true; }
 266
 267  private:
 268   const int thread_count_;
 269   HANDLE *handles_;
 270   std::vector<int> assignment_history_;  // Number of assignment per worker.
 271   std::vector<int> completion_history_;  // Number of completions per worker.
 272   int thread_started_counter_;  // Used to issue unique id to workers.
 273   int shutdown_task_count_;  // Number of tasks told to shutdown
 274   int task_count_;  // Number of assignment tasks waiting to be processed.
 275   NaCl::TimeDelta worker_delay_;  // Time each task takes to complete.
 276   bool allow_help_requests_;  // Workers can signal more workers.
 277   bool shutdown_;  // Set when threads need to terminate.
 278   bool performance_;  // performance vs fairness in thread waking and waiting.
 279 };
 280
 281 // The remaining tests involve several threads with a task to perform
 282 // as directed in the preceding class WorkQueue.
 283 // The task is to:
 284 // a) make sure there are more tasks (there is a task counter).
 285 //    a1) Wait on condition variable if there are no tasks currently.
 286 // b) Call a function to see what should be done.
 287 // c) Do some computation based on the number of milliseconds returned in (b).
 288 // d) go back to (a).
 289
 290 // worker_process() implements the above task for all threads.
 291 // It calls the controlling object to tell the creator about progress,
 292 // and to ask about tasks.
 293 DWORD WINAPI worker_process(void* p) {
 294   NaCl::Lock private_lock;  // Used to waste time on "our work".
 295   int thread_id;
 296
 297   class WorkQueue* queue = reinterpret_cast<WorkQueue*>(p);
 298   {
 299     NaCl::AutoLock auto_lock(*queue->lock());
 300     thread_id = queue->GetThreadId();
 301     if (queue->EveryIdWasAllocated())
 302       queue->all_threads_have_ids_.Signal();  // Tell creator we're ready.
 303   }
 304
 305   while (1) {  // This is the main consumer loop.
 306     NaCl::TimeDelta work_time;
 307     bool could_use_help;
 308     {
 309       NaCl::AutoLock auto_lock(*queue->lock());
 310       if (queue->performance()) {
 311         // Get best performance by not waiting until there is no work.
 312         while (0 == queue->GetRemainingTaskCount() && !queue->shutdown()) {
 313           queue->work_is_available_.Wait(*queue->lock());
 314         }
 315       } else {
 316         // Be ridiculously "fair" and try to let other threads go. :-/.
 317         do {
 318           queue->work_is_available_.Wait(*queue->lock());
 319         } while (0 == queue->GetRemainingTaskCount() && !queue->shutdown());
 320       }
 321       if (queue->shutdown()) {
 322         queue->thread_shutting_down();
 323         return 0;  // termination time.
 324       }
 325       work_time = queue->GetAnAssignment(thread_id);
 326       could_use_help = (queue->GetRemainingTaskCount() > 0) &&
 327           queue->GetAllowHelp();
 328     }  // Release lock
 329
 330     // Do work (out of true critical region, consisting of waiting :-).
 331     if (could_use_help)
 332       queue->work_is_available_.Signal();  // Get help from other threads.
 333
 334     if (work_time > NaCl::TimeDelta::FromMilliseconds(0)) {
 335       NaCl::AutoLock auto_lock(private_lock);
 336       NaCl::ConditionVariable private_cv;
 337       private_cv.TimedWaitRel(private_lock, work_time);
 338     }
 339
 340     {
 341       NaCl::AutoLock auto_lock(*queue->lock());
 342       queue->WorkIsCompleted(thread_id);
 343     }
 344   }
 345 }
 346
 347 #if 0
 348 /* TODO - This test fails periodically, fix it */
 349 TEST(ConditionVariableTest, MultiThreadConsumerTest) {
 350   const int     kThreadCount = 10;
 351   WorkQueue     queue(kThreadCount);  // Start the threads.
 352
 353   NaCl::Lock private_lock;  // Used locally for master to wait.
 354   NaCl::AutoLock private_held_lock(private_lock);
 355   NaCl::ConditionVariable private_cv;
 356
 357   const NaCl::TimeDelta ZERO_MS = NaCl::TimeDelta::FromMilliseconds(0);
 358   const NaCl::TimeDelta TEN_MS = NaCl::TimeDelta::FromMilliseconds(10);
 359   const NaCl::TimeDelta THIRTY_MS = NaCl::TimeDelta::FromMilliseconds(30);
 360   const NaCl::TimeDelta FORTY_FIVE_MS = NaCl::TimeDelta::FromMilliseconds(45);
 361   const NaCl::TimeDelta SIXTY_MS = NaCl::TimeDelta::FromMilliseconds(60);
 362   const NaCl::TimeDelta ONE_HUNDRED_MS = NaCl::TimeDelta::FromMilliseconds(100);
 363
 364   {
 365     NaCl::AutoLock auto_lock(*queue.lock());
 366     while (!queue.EveryIdWasAllocated())
 367       queue.all_threads_have_ids_.Wait(*queue.lock());
 368   }
 369
 370   // Wait a bit more to allow threads to reach their wait state.
 371   private_cv.TimedWaitRel(private_lock, TEN_MS);
 372
 373   {
 374     // Since we have no tasks, all threads should be waiting by now.
 375     NaCl::AutoLock auto_lock(*queue.lock());
 376     EXPECT_EQ(0, queue.GetNumThreadsTakingAssignments());
 377     EXPECT_EQ(0, queue.GetNumThreadsCompletingTasks());
 378     EXPECT_EQ(0, queue.GetRemainingTaskCount());
 379     EXPECT_EQ(0, queue.GetMaxCompletionsByWorkerThread());
 380     EXPECT_EQ(0, queue.GetMinCompletionsByWorkerThread());
 381     EXPECT_EQ(0, queue.GetNumberOfCompletedTasks());
 382
 383     // Set up to make one worker do 3 30ms tasks.
 384     queue.ResetHistory();
 385     queue.SetTaskCount(3);
 386     queue.SetWorkTime(THIRTY_MS);
 387     queue.SetAllowHelp(false);
 388   }
 389   queue.work_is_available()->Signal();  // Start up one thread.
 390   // Wait to allow solo worker insufficient time to get done.
 391   // Should take about 90 ms.
 392   private_cv.TimedWaitRel(private_lock, FORTY_FIVE_MS);
 393
 394   {
 395     // Check that all work HASN'T completed yet.
 396     NaCl::AutoLock auto_lock(*queue.lock());
 397     EXPECT_EQ(1, queue.GetNumThreadsTakingAssignments());
 398     EXPECT_EQ(1, queue.GetNumThreadsCompletingTasks());
 399     EXPECT_GT(2, queue.GetRemainingTaskCount());  // 2 should have started.
 400     EXPECT_GT(3, queue.GetMaxCompletionsByWorkerThread());
 401     EXPECT_EQ(0, queue.GetMinCompletionsByWorkerThread());
 402     EXPECT_EQ(1, queue.GetNumberOfCompletedTasks());
 403   }
 404   // Wait to allow solo workers to get done.
 405   private_cv.TimedWaitRel(private_lock, SIXTY_MS);  // Should take about 45ms more.
 406
 407   {
 408     // Check that all work was done by one thread id.
 409     NaCl::AutoLock auto_lock(*queue.lock());
 410     EXPECT_EQ(1, queue.GetNumThreadsTakingAssignments());
 411     EXPECT_EQ(1, queue.GetNumThreadsCompletingTasks());
 412     EXPECT_EQ(0, queue.GetRemainingTaskCount());
 413     EXPECT_EQ(3, queue.GetMaxCompletionsByWorkerThread());
 414     EXPECT_EQ(0, queue.GetMinCompletionsByWorkerThread());
 415     EXPECT_EQ(3, queue.GetNumberOfCompletedTasks());
 416
 417     // Set up to make each task include getting help from another worker.
 418     queue.ResetHistory();
 419     queue.SetTaskCount(3);
 420     queue.SetWorkTime(THIRTY_MS);
 421     queue.SetAllowHelp(true);
 422   }
 423   queue.work_is_available()->Signal();  // But each worker can signal another.
 424   // Wait to allow the 3 workers to get done.
 425   // Should  take about 30 ms.
 426   private_cv.TimedWaitRel(private_lock, FORTY_FIVE_MS);
 427   {
 428     NaCl::AutoLock auto_lock(*queue.lock());
 429     EXPECT_EQ(3, queue.GetNumThreadsTakingAssignments());
 430     EXPECT_EQ(3, queue.GetNumThreadsCompletingTasks());
 431     EXPECT_EQ(0, queue.GetRemainingTaskCount());
 432     EXPECT_EQ(1, queue.GetMaxCompletionsByWorkerThread());
 433     EXPECT_EQ(0, queue.GetMinCompletionsByWorkerThread());
 434     EXPECT_EQ(3, queue.GetNumberOfCompletedTasks());
 435
 436     // Try to ask all workers to help, and only a few will do the work.
 437     queue.ResetHistory();
 438     queue.SetTaskCount(3);
 439     queue.SetWorkTime(THIRTY_MS);
 440     queue.SetAllowHelp(false);
 441   }
 442   queue.work_is_available()->Broadcast();  // Make them all try.
 443   // Wait to allow the 3 workers to get done.
 444   private_cv.TimedWaitRel(private_lock, FORTY_FIVE_MS);
 445
 446   {
 447     NaCl::AutoLock auto_lock(*queue.lock());
 448     EXPECT_EQ(3, queue.GetNumThreadsTakingAssignments());
 449     EXPECT_EQ(3, queue.GetNumThreadsCompletingTasks());
 450     EXPECT_EQ(0, queue.GetRemainingTaskCount());
 451     EXPECT_EQ(1, queue.GetMaxCompletionsByWorkerThread());
 452     EXPECT_EQ(0, queue.GetMinCompletionsByWorkerThread());
 453     EXPECT_EQ(3, queue.GetNumberOfCompletedTasks());
 454
 455     // Set up to make each task get help from another worker.
 456     queue.ResetHistory();
 457     queue.SetTaskCount(3);
 458     queue.SetWorkTime(THIRTY_MS);
 459     queue.SetAllowHelp(true);  // Allow (unnecessary) help requests.
 460   }
 461   queue.work_is_available()->Broadcast();  // We already signal all threads.
 462   // Wait to allow the 3 workers to get done.
 463   private_cv.TimedWaitRel(private_lock, ONE_HUNDRED_MS);
 464
 465   {
 466     NaCl::AutoLock auto_lock(*queue.lock());
 467     EXPECT_EQ(3, queue.GetNumThreadsTakingAssignments());
 468     EXPECT_EQ(3, queue.GetNumThreadsCompletingTasks());
 469     EXPECT_EQ(0, queue.GetRemainingTaskCount());
 470     EXPECT_EQ(1, queue.GetMaxCompletionsByWorkerThread());
 471     EXPECT_EQ(0, queue.GetMinCompletionsByWorkerThread());
 472     EXPECT_EQ(3, queue.GetNumberOfCompletedTasks());
 473
 474     // Set up to make each task get help from another worker.
 475     queue.ResetHistory();
 476     queue.SetTaskCount(20);
 477     queue.SetWorkTime(THIRTY_MS);
 478     queue.SetAllowHelp(true);
 479   }
 480   queue.work_is_available()->Signal();  // But each worker can signal another.
 481   // Wait to allow the 10 workers to get done.
 482   // Should take about 60 ms.
 483   private_cv.TimedWaitRel(private_lock, ONE_HUNDRED_MS);
 484
 485   {
 486     NaCl::AutoLock auto_lock(*queue.lock());
 487     EXPECT_EQ(10, queue.GetNumThreadsTakingAssignments());
 488     EXPECT_EQ(10, queue.GetNumThreadsCompletingTasks());
 489     EXPECT_EQ(0, queue.GetRemainingTaskCount());
 490     EXPECT_EQ(2, queue.GetMaxCompletionsByWorkerThread());
 491     EXPECT_EQ(2, queue.GetMinCompletionsByWorkerThread());
 492     EXPECT_EQ(20, queue.GetNumberOfCompletedTasks());
 493
 494     // Same as last test, but with Broadcast().
 495     queue.ResetHistory();
 496     queue.SetTaskCount(20);  // 2 tasks per process.
 497     queue.SetWorkTime(THIRTY_MS);
 498     queue.SetAllowHelp(true);
 499   }
 500   queue.work_is_available()->Broadcast();
 501   // Wait to allow the 10 workers to get done.
 502   // Should take about 60 ms.
 503   private_cv.TimedWaitRel(private_lock, ONE_HUNDRED_MS);
 504
 505   {
 506     NaCl::AutoLock auto_lock(*queue.lock());
 507     EXPECT_EQ(10, queue.GetNumThreadsTakingAssignments());
 508     EXPECT_EQ(10, queue.GetNumThreadsCompletingTasks());
 509     EXPECT_EQ(0, queue.GetRemainingTaskCount());
 510     EXPECT_EQ(2, queue.GetMaxCompletionsByWorkerThread());
 511     EXPECT_EQ(2, queue.GetMinCompletionsByWorkerThread());
 512     EXPECT_EQ(20, queue.GetNumberOfCompletedTasks());
 513
 514     queue.SetShutdown();
 515   }
 516   queue.work_is_available()->Broadcast();  // Force check for shutdown.
 517
 518   SPIN_FOR_TIMEDELTA_OR_UNTIL_TRUE(NaCl::TimeDelta::FromMinutes(1),
 519                                   queue.shutdown_task_count() == kThreadCount);
 520   Sleep(10);  // Be sure they're all shutdown.
 521 }
 522
 523 #endif /* disabled MultiThreadConsumerTest */
 524
 525 TEST(ConditionVariableTest, LargeFastTaskTest) {
 526   const int kThreadCount = 200;
 527   WorkQueue queue(kThreadCount);  // Start the threads.
 528
 529   NaCl::Lock private_lock;  // Used locally for master to wait.
 530   NaCl::AutoLock private_held_lock(private_lock);
 531   NaCl::ConditionVariable private_cv;
 532
 533   const NaCl::TimeDelta ZERO_MS = NaCl::TimeDelta::FromMilliseconds(0);
 534   const NaCl::TimeDelta TEN_MS = NaCl::TimeDelta::FromMilliseconds(10);
 535   const NaCl::TimeDelta THIRTY_MS = NaCl::TimeDelta::FromMilliseconds(30);
 536   const NaCl::TimeDelta FORTY_FIVE_MS = NaCl::TimeDelta::FromMilliseconds(45);
 537   const NaCl::TimeDelta SIXTY_MS = NaCl::TimeDelta::FromMilliseconds(60);
 538   const NaCl::TimeDelta ONE_HUNDRED_MS = NaCl::TimeDelta::FromMilliseconds(100);
 539
 540   {
 541     NaCl::AutoLock auto_lock(*queue.lock());
 542     while (!queue.EveryIdWasAllocated())
 543       queue.all_threads_have_ids_.Wait(*queue.lock());
 544   }
 545
 546   // Wait a bit more to allow threads to reach their wait state.
 547   private_cv.TimedWaitRel(private_lock, THIRTY_MS);
 548
 549   {
 550     // Since we have no tasks, all threads should be waiting by now.
 551     NaCl::AutoLock auto_lock(*queue.lock());
 552     EXPECT_EQ(0, queue.GetNumThreadsTakingAssignments());
 553     EXPECT_EQ(0, queue.GetNumThreadsCompletingTasks());
 554     EXPECT_EQ(0, queue.GetRemainingTaskCount());
 555     EXPECT_EQ(0, queue.GetMaxCompletionsByWorkerThread());
 556     EXPECT_EQ(0, queue.GetMinCompletionsByWorkerThread());
 557     EXPECT_EQ(0, queue.GetNumberOfCompletedTasks());
 558
 559     // Set up to make all workers do (an average of) 20 tasks.
 560     queue.ResetHistory();
 561     queue.SetTaskCount(20 * kThreadCount);
 562     queue.SetWorkTime(FORTY_FIVE_MS);
 563     queue.SetAllowHelp(false);
 564   }
 565   queue.work_is_available()->Broadcast();  // Start up all threads.
 566   // Wait until we've handed out all tasks.
 567   {
 568     NaCl::AutoLock auto_lock(*queue.lock());
 569     while (queue.GetRemainingTaskCount() != 0)
 570       queue.no_more_tasks()->Wait(*queue.lock());
 571   }
 572
 573   // Wait till the last of the tasks complete.
 574   // Don't bother to use locks: We may not get info in time... but we'll see it
 575   // eventually.
 576   SPIN_FOR_TIMEDELTA_OR_UNTIL_TRUE(NaCl::TimeDelta::FromMinutes(1),
 577                                     20 * kThreadCount ==
 578                                       queue.GetNumberOfCompletedTasks());
 579
 580   {
 581     // With Broadcast(), every thread should have participated.
 582     // but with racing.. they may not all have done equal numbers of tasks.
 583     NaCl::AutoLock auto_lock(*queue.lock());
 584     EXPECT_EQ(kThreadCount, queue.GetNumThreadsTakingAssignments());
 585     EXPECT_EQ(kThreadCount, queue.GetNumThreadsCompletingTasks());
 586     EXPECT_EQ(0, queue.GetRemainingTaskCount());
 587     EXPECT_LE(20, queue.GetMaxCompletionsByWorkerThread());
 588     // EXPECT_LE(1, queue.GetMinCompletionsByWorkerThread());
 589     EXPECT_EQ(20 * kThreadCount, queue.GetNumberOfCompletedTasks());
 590
 591     // Set up to make all workers do (an average of) 4 tasks.
 592     queue.ResetHistory();
 593     queue.SetTaskCount(kThreadCount * 4);
 594     queue.SetWorkTime(FORTY_FIVE_MS);
 595     queue.SetAllowHelp(true);  // Might outperform Broadcast().
 596   }
 597   queue.work_is_available()->Signal();  // Start up one thread.
 598
 599   // Wait until we've handed out all tasks
 600   {
 601     NaCl::AutoLock auto_lock(*queue.lock());
 602     while (queue.GetRemainingTaskCount() != 0)
 603       queue.no_more_tasks()->Wait(*queue.lock());
 604   }
 605
 606   // Wait till the last of the tasks complete.
 607   // Don't bother to use locks: We may not get info in time... but we'll see it
 608   // eventually.
 609   SPIN_FOR_TIMEDELTA_OR_UNTIL_TRUE(NaCl::TimeDelta::FromMinutes(1),
 610                                     4 * kThreadCount ==
 611                                       queue.GetNumberOfCompletedTasks());
 612
 613   {
 614     // With Signal(), every thread should have participated.
 615     // but with racing.. they may not all have done four tasks.
 616     NaCl::AutoLock auto_lock(*queue.lock());
 617     EXPECT_EQ(kThreadCount, queue.GetNumThreadsTakingAssignments());
 618     EXPECT_EQ(kThreadCount, queue.GetNumThreadsCompletingTasks());
 619     EXPECT_EQ(0, queue.GetRemainingTaskCount());
 620     EXPECT_LE(4, queue.GetMaxCompletionsByWorkerThread());
 621     // EXPECT_LE(1, queue.GetMinCompletionsByWorkerThread());
 622     EXPECT_EQ(4 * kThreadCount, queue.GetNumberOfCompletedTasks());
 623
 624     queue.SetShutdown();
 625   }
 626   queue.work_is_available()->Broadcast();  // Force check for shutdown.
 627
 628   // Wait for shutdows to complete.
 629   SPIN_FOR_TIMEDELTA_OR_UNTIL_TRUE(NaCl::TimeDelta::FromMinutes(1),
 630                                     queue.shutdown_task_count() ==
 631                                       kThreadCount);
 632   Sleep(10);  // Be sure they're all shutdown.
 633 }
 634
 635 #endif  // NACL_WINDOWS
 636
 637 }  // namespace