Flush SequenceWorkerPool tasks after each unit test.

base/threading/sequenced_worker_pool.cc

 // 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.
 
 #include "base/threading/sequenced_worker_pool.h"
 
 #include <list>
 #include <map>
 #include <set>
 #include <utility>
@@ skipping 371 matching lines @@
   // This lock protects |everything in this class|. Do not read or modify
   // anything without holding this lock. Do not block while holding this
   // lock.
   mutable Lock lock_;
 
   // Condition variable that is waited on by worker threads until new
   // tasks are posted or shutdown starts.
   ConditionVariable has_work_cv_;
 
   // Condition variable that is waited on by non-worker threads (in
-  // FlushForTesting()) until IsIdle() goes to true.
+  // FlushForTesting()) until IsIdle() goes to true or Shutdown completes.

[michaeln, 2012/12/21 03:17:55]

I might be able to better normalize things by introducing a shutdown_complete_
bool + cv pair but I think the smaller changes here are enough to break the
deadlock... i'll see what the trybots think.

   ConditionVariable is_idle_cv_;
 
   // Condition variable that is waited on by non-worker threads (in
   // Shutdown()) until CanShutdown() goes to true.
   ConditionVariable can_shutdown_cv_;
 
   // The maximum number of worker threads we'll create.
   const size_t max_threads_;
 
   const std::string thread_name_prefix_;
@@ skipping 200 matching lines @@
 bool SequencedWorkerPool::Inner::IsRunningSequenceOnCurrentThread(
     SequenceToken sequence_token) const {
   AutoLock lock(lock_);
   ThreadMap::const_iterator found = threads_.find(PlatformThread::CurrentId());
   if (found == threads_.end())
     return false;
   return found->second->running_sequence().Equals(sequence_token);
 }
 
 void SequencedWorkerPool::Inner::FlushForTesting() {
+  base::ThreadRestrictions::ScopedAllowWait allow_wait;
   AutoLock lock(lock_);
-  while (!IsIdle())
+  while (!IsIdle() && !shutdown_called_)
     is_idle_cv_.Wait();
 }
 
 void SequencedWorkerPool::Inner::SignalHasWorkForTesting() {
   SignalHasWork();
 }
 
 void SequencedWorkerPool::Inner::Shutdown(
     int max_new_blocking_tasks_after_shutdown) {
   DCHECK_GE(max_new_blocking_tasks_after_shutdown, 0);
   {
     AutoLock lock(lock_);
 
     if (shutdown_called_)
       return;
     shutdown_called_ = true;
     max_blocking_tasks_after_shutdown_ = max_new_blocking_tasks_after_shutdown;
 
     // Tickle the threads. This will wake up a waiting one so it will know that
     // it can exit, which in turn will wake up any other waiting ones.
     SignalHasWork();
 
     // There are no pending or running tasks blocking shutdown, we're done.
-    if (CanShutdown())
+    if (CanShutdown()) {
+      is_idle_cv_.Signal();

[michaeln, 2012/12/21 02:50:47]

FlushForTesting can wait forever if Shutdown happens after its wait has begun.
So we signal upon return to wake them.

Some existing tests are disabled due to this problem, and when adding my
additional calls to FlushForTesting more test timeouts show up.

       return;
+    }
   }
 
   // If we're here, then something is blocking shutdown.  So wait for
   // CanShutdown() to go to true.
 
   if (testing_observer_)
     testing_observer_->WillWaitForShutdown();
 
   TimeTicks shutdown_wait_begin = TimeTicks::Now();
 
   {
     base::ThreadRestrictions::ScopedAllowWait allow_wait;
     AutoLock lock(lock_);
     while (!CanShutdown())
       can_shutdown_cv_.Wait();
+    is_idle_cv_.Signal();
   }
   UMA_HISTOGRAM_TIMES("SequencedWorkerPool.ShutdownDelayTime",
                       TimeTicks::Now() - shutdown_wait_begin);
 }
 
 void SequencedWorkerPool::Inner::ThreadLoop(Worker* this_worker) {
   {
     AutoLock lock(lock_);
     DCHECK(thread_being_created_);
     thread_being_created_ = false;
@@ skipping 487 matching lines @@
 void SequencedWorkerPool::SignalHasWorkForTesting() {
   inner_->SignalHasWorkForTesting();
 }
 
 void SequencedWorkerPool::Shutdown(int max_new_blocking_tasks_after_shutdown) {
   DCHECK(constructor_message_loop_->BelongsToCurrentThread());
   inner_->Shutdown(max_new_blocking_tasks_after_shutdown);
 }
 
 }  // namespace base