Flush SequenceWorkerPool tasks after each unit test.

base/threading/sequenced_worker_pool.cc

Index: base/threading/sequenced_worker_pool.cc
===================================================================
--- base/threading/sequenced_worker_pool.cc	(revision 176718)
+++ base/threading/sequenced_worker_pool.cc	(working copy)
@@ -281,7 +281,7 @@
 
   bool IsRunningSequenceOnCurrentThread(SequenceToken sequence_token) const;
 
-  void FlushForTesting();
+  void CleanupForTesting();
 
   void SignalHasWorkForTesting();
 
@@ -299,9 +299,12 @@
     GET_WORK_WAIT,
   };
 
-  // Returns whether there are no more pending tasks and all threads
-  // are idle.  Must be called under lock.
-  bool IsIdle() const;
+  enum CleanupState {
+    CLEANUP_REQUESTED,
+    CLEANUP_WAITING,
+    CLEANUP_RUNNING,
+    CLEANUP_DONE
+  };
 
   // Called from within the lock, this converts the given token name into a
   // token ID, creating a new one if necessary.
@@ -334,6 +337,8 @@
                         TimeDelta* wait_time,
                         std::vector<Closure>* delete_these_outside_lock);
 
+  void HandleCleanup();
+
   // Peforms init and cleanup around running the given task. WillRun...
   // returns the value from PrepareToStartAdditionalThreadIfNecessary.
   // The calling code should call FinishStartingAdditionalThread once the
@@ -389,10 +394,6 @@
   ConditionVariable has_work_cv_;
 
   // Condition variable that is waited on by non-worker threads (in
-  // FlushForTesting()) until IsIdle() goes to true.
-  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_;
 
@@ -450,6 +451,11 @@
   // has been called.
   int max_blocking_tasks_after_shutdown_;
 
+  // State used to cleanup for testing, all guarded by lock_.
+  CleanupState cleanup_state_;
+  size_t cleanup_idlers_;
+  ConditionVariable cleanup_cv_;
+
   TestingObserver* const testing_observer_;
 
   DISALLOW_COPY_AND_ASSIGN(Inner);
@@ -493,7 +499,6 @@
       last_sequence_number_(0),
       lock_(),
       has_work_cv_(&lock_),
-      is_idle_cv_(&lock_),
       can_shutdown_cv_(&lock_),
       max_threads_(max_threads),
       thread_name_prefix_(thread_name_prefix),
@@ -505,6 +510,9 @@
       trace_id_(0),
       shutdown_called_(false),
       max_blocking_tasks_after_shutdown_(0),
+      cleanup_state_(CLEANUP_DONE),
+      cleanup_idlers_(0),
+      cleanup_cv_(&lock_),
       testing_observer_(observer) {}
 
 SequencedWorkerPool::Inner::~Inner() {
@@ -609,10 +617,24 @@
   return found->second->running_sequence().Equals(sequence_token);
 }
 
-void SequencedWorkerPool::Inner::FlushForTesting() {
+// See https://code.google.com/p/chromium/issues/detail?id=168415
+// * executes all pending non-delayed tasks and deletes the associated closures
+// * skips executing all pending delayed tasks, but does delete their closures
+// * accepts new tasks while in the act of cleaning up
+void SequencedWorkerPool::Inner::CleanupForTesting() {
+  DCHECK(!RunsTasksOnCurrentThread());
+  base::ThreadRestrictions::ScopedAllowWait allow_wait;
   AutoLock lock(lock_);
-  while (!IsIdle())
-    is_idle_cv_.Wait();
+  if (shutdown_called_)
+    return;
+  CHECK_EQ(CLEANUP_DONE, cleanup_state_);
+  if (!thread_being_created_ && threads_.empty())
+    return;
+  cleanup_state_ = CLEANUP_REQUESTED;
+  cleanup_idlers_ = 0;
+  has_work_cv_.Signal();
+  while (cleanup_state_ != CLEANUP_DONE)
+    cleanup_cv_.Wait();
 }
 
 void SequencedWorkerPool::Inner::SignalHasWorkForTesting() {
@@ -624,9 +646,9 @@
   DCHECK_GE(max_new_blocking_tasks_after_shutdown, 0);
   {
     AutoLock lock(lock_);
-
     if (shutdown_called_)
       return;
+    CHECK_EQ(CLEANUP_DONE, cleanup_state_);
     shutdown_called_ = true;
     max_blocking_tasks_after_shutdown_ = max_new_blocking_tasks_after_shutdown;
 
@@ -672,6 +694,9 @@
       base::mac::ScopedNSAutoreleasePool autorelease_pool;
 #endif
 
+      if (cleanup_state_ != CLEANUP_DONE)
+        HandleCleanup();

[jar (doing other things), 2013/01/15 22:59:59]

It is surprising we call this in 3 of the 4 states.

Probably more readable to do all the state checking in the HandleCleanup().

+
       // See GetWork for what delete_these_outside_lock is doing.
       SequencedTask task;
       TimeDelta wait_time;
@@ -717,6 +742,20 @@
           task.task = Closure();
         }
         DidRunWorkerTask(task);  // Must be done inside the lock.
+      } else if (cleanup_state_ == CLEANUP_RUNNING) {
+        switch (status) {
+          case GET_WORK_NOT_FOUND:
+            cleanup_state_ = CLEANUP_DONE;
+            cleanup_cv_.Broadcast();
+            break;
+          case GET_WORK_WAIT: {
+              AutoUnlock unlock(lock_);
+              delete_these_outside_lock.clear();
+            }
+            break;
+          default:
+            NOTREACHED();
+        }
       } else {
         // When we're terminating and there's no more work, we can
         // shut down, other workers can complete any pending or new tasks.
@@ -730,9 +769,6 @@
             blocking_shutdown_pending_task_count_ == 0)
           break;
         waiting_thread_count_++;
-        // This is the only time that IsIdle() can go to true.
-        if (IsIdle())
-          is_idle_cv_.Signal();
 
         switch (status) {
           case GET_WORK_NOT_FOUND:
@@ -757,9 +793,28 @@
   can_shutdown_cv_.Signal();
 }
 
-bool SequencedWorkerPool::Inner::IsIdle() const {
+void SequencedWorkerPool::Inner::HandleCleanup() {
   lock_.AssertAcquired();
-  return pending_tasks_.empty() && waiting_thread_count_ == threads_.size();
+  if (cleanup_state_ == CLEANUP_REQUESTED) {
+    // We win! We get to do the cleanup as soon as the others wise up and idle.
+    cleanup_state_ = CLEANUP_WAITING;
+    while (thread_being_created_ &&
+           cleanup_idlers_ != threads_.size() - 1) {
+      has_work_cv_.Signal();
+      cleanup_cv_.Wait();
+    }
+    cleanup_state_ = CLEANUP_RUNNING;
+    return;
+  }
+  if (cleanup_state_ == CLEANUP_WAITING) {
+    // Another worker thread is cleaning up, we idle here until thats done.
+    ++cleanup_idlers_;
+    cleanup_cv_.Broadcast();
+    while (cleanup_state_ != CLEANUP_DONE)
+      cleanup_cv_.Wait();
+    return;
+  }
+  DCHECK_EQ(CLEANUP_RUNNING, cleanup_state_);
 }
 
 int SequencedWorkerPool::Inner::LockedGetNamedTokenID(
@@ -838,10 +893,26 @@
       continue;
     }
 
+    // Let's try running all delayed tasks during cleanup?

[jar (doing other things), 2013/01/15 22:59:59]

What's up with the question mark?

...also... I thought you gave up on running them, and were going to delete them.

+    if (cleanup_state_ == CLEANUP_RUNNING) {
+      *task = *i;
+      pending_tasks_.erase(i);
+      if (task->shutdown_behavior == BLOCK_SHUTDOWN)
+        blocking_shutdown_pending_task_count_--;
+      status = GET_WORK_FOUND;
+      break;
+    }
+
     if (i->time_to_run > current_time) {
       // The time to run has not come yet.
       *wait_time = i->time_to_run - current_time;
       status = GET_WORK_WAIT;
+      /*
+      if (cleanup_state_ == CLEANUP_RUNNING) {
+        delete_these_outside_lock->push_back(i->task);
+        pending_tasks_.erase(i);
+      }
+      */
       break;
     }
 
@@ -1146,7 +1217,7 @@
 }
 
 void SequencedWorkerPool::FlushForTesting() {
-  inner_->FlushForTesting();
+  inner_->CleanupForTesting();
 }
 
 void SequencedWorkerPool::SignalHasWorkForTesting() {