Reduce flakiness of condition var unit test

base/condition_variable_unittest.cc

 // Copyright (c) 2006-2008 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 <time.h>
 #include <algorithm>
 #include <vector>
 
 #include "base/condition_variable.h"
 #include "base/logging.h"
 #include "base/platform_thread.h"
 #include "base/scoped_ptr.h"
 #include "base/spin_wait.h"
 #include "base/thread_collision_warner.h"
+#include "base/time.h"
 #include "testing/gtest/include/gtest/gtest.h"
 #include "testing/platform_test.h"
 
 using base::TimeDelta;
 using base::TimeTicks;
 
 namespace {
 //------------------------------------------------------------------------------
 // Define our test class, with several common variables.
 //------------------------------------------------------------------------------
@@ skipping 64 matching lines @@
 
   //----------------------------------------------------------------------------
   // 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;
+  TimeDelta GetWorkTime() const;
 
   void SetWorkTime(TimeDelta delay);
   void SetTaskCount(int count);
   void SetAllowHelp(bool allow);
 
   // 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
@@ skipping 62 matching lines @@
 
   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();
 }
 
-// This test is flaky due to excessive timing sensitivity.
-// http://code.google.com/p/chromium/issues/detail?id=3599
-// TODO(jar): A recent change to the WorkQueue fixed flakyness for the test
-// LargeFastTaskTest.  Specifically, the test was accessing the member variable
-// WorkQueue::shutdown_task_count_ without a lock held.  The MultiThreadConsumer
-// test now ran successfully for 500 times on RalphL's machine.  RalphL did not
-// want to blindly re-enable this test if you know of other issues with it, but
-// it appears that the flakyness is gone, but I'll leave that to you to verify.
-TEST_F(ConditionVariableTest, DISABLED_MultiThreadConsumerTest) {
+// Test serial task servicing, as well as two parallel task servicing methods.
+TEST_F(ConditionVariableTest, MultiThreadConsumerTest) {
   const int kThreadCount = 10;
   WorkQueue queue(kThreadCount);  // Start the threads.
 
-  Lock private_lock;  // Used locally for master to wait.
-  AutoLock private_held_lock(private_lock);
-  ConditionVariable private_cv(&private_lock);
+  const int kTaskCount = 10;  // Number of tasks in each mini-test here.
+
+  base::Time start_time;  // Used to time task processing.
 
   {
     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.
-  private_cv.TimedWait(kTenMs);
+  // If threads aren't in a wait state, they may start to gobble up tasks in
+  // parallel, short-circuiting (breaking) this test.
+  PlatformThread::Sleep(100);
 
   {
-    // Since we have no tasks, all threads should be waiting by now.
+    // Since we have no tasks yet, all threads should be waiting by now.
     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 one worker do 3 30ms tasks.
+    // Set up to make one worker do 30ms tasks sequentially.
     queue.ResetHistory();
-    queue.SetTaskCount(3);
+    queue.SetTaskCount(kTaskCount);
     queue.SetWorkTime(kThirtyMs);
     queue.SetAllowHelp(false);
+
+    start_time = base::Time::Now();
   }
+
   queue.work_is_available()->Signal();  // Start up one thread.
-  // Wait to allow solo worker insufficient time to get done.
-  private_cv.TimedWait(kFortyFiveMs);  // Should take about 90 ms.
+
 
   {
-    // Check that all work HASN'T completed yet.
+    // Wait until all 10 work tasks have at least been assigned.
     AutoLock auto_lock(*queue.lock());
+    while(queue.task_count())
+      queue.no_more_tasks()->Wait();
+    // The last of the tasks *might* still be running, but... all but one should
+    // be done by now, since tasks are being done serially.
+    EXPECT_LT(queue.GetWorkTime().InMilliseconds() * (kTaskCount - 1),
+              (base::Time::Now() - start_time).InMilliseconds());
+
     EXPECT_EQ(1, queue.GetNumThreadsTakingAssignments());
     EXPECT_EQ(1, queue.GetNumThreadsCompletingTasks());
-    EXPECT_GT(2, queue.task_count());  // 2 should have started.
-    EXPECT_GT(3, queue.GetMaxCompletionsByWorkerThread());
+    EXPECT_LE(kTaskCount - 1, queue.GetMaxCompletionsByWorkerThread());
     EXPECT_EQ(0, queue.GetMinCompletionsByWorkerThread());
-    EXPECT_EQ(1, queue.GetNumberOfCompletedTasks());
+    EXPECT_LE(kTaskCount - 1, queue.GetNumberOfCompletedTasks());
   }
-  // Wait to allow solo workers to get done.
-  private_cv.TimedWait(kSixtyMs);  // Should take about 45ms more.
+
+  // Wait to be sure all tasks are done.
+  while (1) {
+    {
+      AutoLock auto_lock(*queue.lock());
+      if (kTaskCount == queue.GetNumberOfCompletedTasks())
+        break;
+    }
+    PlatformThread::Sleep(30);  // Wait a little.
+  }
 
   {
     // Check that all work was done by one thread id.
     AutoLock auto_lock(*queue.lock());
     EXPECT_EQ(1, queue.GetNumThreadsTakingAssignments());
     EXPECT_EQ(1, queue.GetNumThreadsCompletingTasks());
     EXPECT_EQ(0, queue.task_count());
-    EXPECT_EQ(3, queue.GetMaxCompletionsByWorkerThread());
+    EXPECT_EQ(kTaskCount, queue.GetMaxCompletionsByWorkerThread());
     EXPECT_EQ(0, queue.GetMinCompletionsByWorkerThread());
-    EXPECT_EQ(3, queue.GetNumberOfCompletedTasks());
+    EXPECT_EQ(kTaskCount, queue.GetNumberOfCompletedTasks());
 
-    // Set up to make each task include getting help from another worker.
+    // 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(3);
+    queue.SetTaskCount(kTaskCount);
     queue.SetWorkTime(kThirtyMs);
     queue.SetAllowHelp(true);
+
+    start_time = base::Time::Now();
   }
+
   queue.work_is_available()->Signal();  // But each worker can signal another.
-  // Wait to allow the 3 workers to get done.
-  private_cv.TimedWait(kFortyFiveMs);  // Should  take about 30 ms.
+  // Wait to allow the all workers to get done.
+  while (1) {
+    {
+      AutoLock auto_lock(*queue.lock());
+      if (kTaskCount == queue.GetNumberOfCompletedTasks())
+        break;
+    }
+    PlatformThread::Sleep(30);  // Wait a little.
+  }
 
   {
+    // Wait until all work tasks have at least been assigned.
     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());
+    while(queue.task_count())
+      queue.no_more_tasks()->Wait();
+    // Since they can all run almost in parallel, there is no guarantee that all
+    // tasks are finished, but we should have gotten here faster than it would
+    // take to run all tasks serially.
+    EXPECT_GT(queue.GetWorkTime().InMilliseconds() * (kTaskCount - 1),
+              (base::Time::Now() - start_time).InMilliseconds());
+
+    // To avoid racy assumptions, we'll just assert that at least 2 threads
+    // did work.
+    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 to allow the 3 workers to get done.
-  private_cv.TimedWait(kFortyFiveMs);
+  PlatformThread::Sleep(45);
 
   {
     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();  // We already signal all threads.
   // Wait to allow the 3 workers to get done.
-  private_cv.TimedWait(kOneHundredMs);
+  PlatformThread::Sleep(100);
 
   {
     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);
     queue.SetWorkTime(kThirtyMs);
     queue.SetAllowHelp(true);
   }
   queue.work_is_available()->Signal();  // But each worker can signal another.
   // Wait to allow the 10 workers to get done.
-  private_cv.TimedWait(kOneHundredMs);  // Should take about 60 ms.
+  PlatformThread::Sleep(100);  // Should take about 60 ms.
 
   {
     AutoLock auto_lock(*queue.lock());
     EXPECT_EQ(10, queue.GetNumThreadsTakingAssignments());
     EXPECT_EQ(10, queue.GetNumThreadsCompletingTasks());
     EXPECT_EQ(0, queue.task_count());
     EXPECT_EQ(2, queue.GetMaxCompletionsByWorkerThread());
     EXPECT_EQ(2, queue.GetMinCompletionsByWorkerThread());
     EXPECT_EQ(20, queue.GetNumberOfCompletedTasks());
 
     // Same as last test, but with Broadcast().
     queue.ResetHistory();
     queue.SetTaskCount(20);  // 2 tasks per process.
     queue.SetWorkTime(kThirtyMs);
     queue.SetAllowHelp(true);
   }
   queue.work_is_available()->Broadcast();
   // Wait to allow the 10 workers to get done.
-  private_cv.TimedWait(kOneHundredMs);  // Should take about 60 ms.
+  PlatformThread::Sleep(100);  // Should take about 60 ms.
 
   {
     AutoLock auto_lock(*queue.lock());
     EXPECT_EQ(10, queue.GetNumThreadsTakingAssignments());
     EXPECT_EQ(10, queue.GetNumThreadsCompletingTasks());
     EXPECT_EQ(0, queue.task_count());
     EXPECT_EQ(2, queue.GetMaxCompletionsByWorkerThread());
     EXPECT_EQ(2, queue.GetMinCompletionsByWorkerThread());
     EXPECT_EQ(20, queue.GetNumberOfCompletedTasks());
 
@@ skipping 263 matching lines @@
   return count;
 }
 
 int WorkQueue::GetNumberOfCompletedTasks() const {
   int total = 0;
   for (int i = 0; i < thread_count_; ++i)
     total += completion_history_[i];
   return total;
 }
 
+TimeDelta WorkQueue::GetWorkTime() const {
+  return worker_delay_;
+}
+
 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;
@@ skipping 64 matching lines @@
 
     {
       AutoLock auto_lock(lock_);
       // Send notification that we completed our "work."
       WorkIsCompleted(thread_id);
     }
   }
 }
 
 }  // namespace