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 263 matching lines @@
                 SequenceToken sequence_token,
                 WorkerShutdown shutdown_behavior,
                 const tracked_objects::Location& from_here,
                 const Closure& task,
                 TimeDelta delay);
 
   bool RunsTasksOnCurrentThread() const;
 
   bool IsRunningSequenceOnCurrentThread(SequenceToken sequence_token) const;
 
-  void FlushForTesting();
+  void CleanupForTesting();
 
   void SignalHasWorkForTesting();
 
   int GetWorkSignalCountForTesting() const;
 
   void Shutdown(int max_blocking_tasks_after_shutdown);
 
   // Runs the worker loop on the background thread.
   void ThreadLoop(Worker* this_worker);
 
  private:
   enum GetWorkStatus {
     GET_WORK_FOUND,
     GET_WORK_NOT_FOUND,
     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_STARTING,
+    CLEANUP_RUNNING,
+    CLEANUP_FINISHING,
+    CLEANUP_DONE,
+  };
 
   // Called from within the lock, this converts the given token name into a
   // token ID, creating a new one if necessary.
   int LockedGetNamedTokenID(const std::string& name);
 
   // Called from within the lock, this returns the next sequence task number.
   int64 LockedGetNextSequenceTaskNumber();
 
   // Called from within the lock, returns the shutdown behavior of the task
   // running on the currently executing worker thread. If invoked from a thread
@@ skipping 12 matching lines @@
   //    filled in the time to wait until the next task to run. In this case, the
   //    caller should wait the time.
   //
   // In any case, the calling code should clear the given
   // delete_these_outside_lock vector the next time the lock is released.
   // See the implementation for a more detailed description.
   GetWorkStatus GetWork(SequencedTask* task,
                         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
   // lock is released if the return values is nonzero.
   int WillRunWorkerTask(const SequencedTask& task);
   void DidRunWorkerTask(const SequencedTask& task);
 
   // Returns true if there are no threads currently running the given
   // sequence token.
   bool IsSequenceTokenRunnable(int sequence_token_id) const;
@@ skipping 35 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.
-  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_;
 
   // Associates all known sequence token names with their IDs.
   std::map<std::string, int> named_sequence_tokens_;
@@ skipping 37 matching lines @@
   int trace_id_;
 
   // Set when Shutdown is called and no further tasks should be
   // allowed, though we may still be running existing tasks.
   bool shutdown_called_;
 
   // The number of new BLOCK_SHUTDOWN tasks that may be posted after Shudown()
   // 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);
 };
 
 // Worker definitions ---------------------------------------------------------
 
 SequencedWorkerPool::Worker::Worker(
     const scoped_refptr<SequencedWorkerPool>& worker_pool,
     int thread_number,
@@ skipping 23 matching lines @@
 
 SequencedWorkerPool::Inner::Inner(
     SequencedWorkerPool* worker_pool,
     size_t max_threads,
     const std::string& thread_name_prefix,
     TestingObserver* observer)
     : worker_pool_(worker_pool),
       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),
       thread_being_created_(false),
       waiting_thread_count_(0),
       blocking_shutdown_thread_count_(0),
       next_sequence_task_number_(0),
       blocking_shutdown_pending_task_count_(0),
       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() {
   // You must call Shutdown() before destroying the pool.
   DCHECK(shutdown_called_);
 
   // Need to explicitly join with the threads before they're destroyed or else
   // they will be running when our object is half torn down.
   for (ThreadMap::iterator it = threads_.begin(); it != threads_.end(); ++it)
     it->second->Join();
@@ skipping 84 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() {
+// See https://code.google.com/p/chromium/issues/detail?id=168415
+void SequencedWorkerPool::Inner::CleanupForTesting() {
+  DCHECK(!RunsTasksOnCurrentThread());
+  base::ThreadRestrictions::ScopedAllowWait allow_wait;
   AutoLock lock(lock_);
-  while (!IsIdle())
-    is_idle_cv_.Wait();
+  CHECK_EQ(CLEANUP_DONE, cleanup_state_);
+  if (shutdown_called_)
+    return;
+  if (pending_tasks_.empty() && waiting_thread_count_ == threads_.size())
+    return;
+  cleanup_state_ = CLEANUP_REQUESTED;
+  cleanup_idlers_ = 0;
+  has_work_cv_.Signal();
+  while (cleanup_state_ != CLEANUP_DONE)
+    cleanup_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_);
-
+    // Cleanup and Shutdown should not be called concurrently.
+    CHECK_EQ(CLEANUP_DONE, cleanup_state_);
     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.
@@ skipping 27 matching lines @@
     std::pair<ThreadMap::iterator, bool> result =
         threads_.insert(
             std::make_pair(this_worker->tid(), make_linked_ptr(this_worker)));
     DCHECK(result.second);
 
     while (true) {
 #if defined(OS_MACOSX)
       base::mac::ScopedNSAutoreleasePool autorelease_pool;
 #endif
 
+      HandleCleanup();
+
       // See GetWork for what delete_these_outside_lock is doing.
       SequencedTask task;
       TimeDelta wait_time;
       std::vector<Closure> delete_these_outside_lock;
       GetWorkStatus status =
           GetWork(&task, &wait_time, &delete_these_outside_lock);
       if (status == GET_WORK_FOUND) {
         TRACE_EVENT_FLOW_END0("task", "SequencedWorkerPool::PostTask",
             TRACE_ID_MANGLE(GetTaskTraceID(task, static_cast<void*>(this))));
         TRACE_EVENT2("task", "SequencedWorkerPool::ThreadLoop",
@@ skipping 28 matching lines @@
           // same reason we do this with delete_these_oustide_lock.
           // Also, do it before calling set_running_task_info() so
           // that sequence-checking from within the task's destructor
           // still works.
           task.task = Closure();
 
           this_worker->set_running_task_info(
               SequenceToken(), CONTINUE_ON_SHUTDOWN);
         }
         DidRunWorkerTask(task);  // Must be done inside the lock.
+      } else if (cleanup_state_ == CLEANUP_RUNNING) {
+        switch (status) {
+          case GET_WORK_WAIT: {
+              AutoUnlock unlock(lock_);
+              delete_these_outside_lock.clear();
+            }
+            break;
+          case GET_WORK_NOT_FOUND:
+            CHECK(delete_these_outside_lock.empty());
+            cleanup_state_ = CLEANUP_FINISHING;
+            cleanup_cv_.Broadcast();
+            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.
         // We can get additional tasks posted after shutdown_called_ is set
         // but only worker threads are allowed to post tasks at that time, and
         // the workers responsible for posting those tasks will be available
         // to run them. Also, there may be some tasks stuck behind running
         // ones with the same sequence token, but additional threads won't
         // help this case.
         if (shutdown_called_ &&
             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:
             has_work_cv_.Wait();
             break;
           case GET_WORK_WAIT:
             has_work_cv_.TimedWait(wait_time);
             break;
           default:
             NOTREACHED();
         }
         waiting_thread_count_--;
       }
     }
   }  // Release lock_.
 
   // We noticed we should exit. Wake up the next worker so it knows it should
   // exit as well (because the Shutdown() code only signals once).
   SignalHasWork();
 
   // Possibly unblock shutdown.
   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_DONE)
+    return;
+  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_STARTING;
+    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_STARTING) {
+    // Another worker thread is cleaning up, we idle here until thats done.
+    ++cleanup_idlers_;
+    cleanup_cv_.Broadcast();
+    while (cleanup_state_ != CLEANUP_FINISHING) {
+      cleanup_cv_.Wait();
+    }
+    --cleanup_idlers_;
+    cleanup_cv_.Broadcast();
+    return;
+  }
+  if (cleanup_state_ == CLEANUP_FINISHING) {
+    // We wait for all idlers to wake up prior to being DONE.
+    while (cleanup_idlers_ != 0) {
+      cleanup_cv_.Broadcast();
+      cleanup_cv_.Wait();
+    }
+    if (cleanup_state_ == CLEANUP_FINISHING) {
+      cleanup_state_ = CLEANUP_DONE;
+      cleanup_cv_.Signal();
+    }
+    return;
+  }
 }
 
 int SequencedWorkerPool::Inner::LockedGetNamedTokenID(
     const std::string& name) {
   lock_.AssertAcquired();
   DCHECK(!name.empty());
 
   std::map<std::string, int>::const_iterator found =
       named_sequence_tokens_.find(name);
   if (found != named_sequence_tokens_.end())
@@ skipping 83 matching lines @@
       // happen.
       delete_these_outside_lock->push_back(i->task);
       pending_tasks_.erase(i++);
       continue;
     }
 
     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) {
+        // Deferred tasks are deleted when cleaning up, see Inner::ThreadLoop.
+        delete_these_outside_lock->push_back(i->task);
+        pending_tasks_.erase(i);
+      }
       break;
     }
 
     // Found a runnable task.
     *task = *i;
     pending_tasks_.erase(i);
     if (task->shutdown_behavior == BLOCK_SHUTDOWN) {
       blocking_shutdown_pending_task_count_--;
     }
 
@@ skipping 86 matching lines @@
   //     cause a worker to start since one is pending.
   //  3. Main thread initiates shutdown.
   //  4. No more threads are created since the shutdown_called_ flag is set.
   //
   // The result is that one may expect that max_threads_ workers to be created
   // given the workload, but in reality fewer may be created because the
   // sequence of thread creation on the background threads is racing with the
   // shutdown call.
   if (!shutdown_called_ &&
       !thread_being_created_ &&
+      cleanup_state_ == CLEANUP_DONE &&
       threads_.size() < max_threads_ &&
       waiting_thread_count_ == 0) {
     // We could use an additional thread if there's work to be done.
     for (PendingTaskSet::const_iterator i = pending_tasks_.begin();
          i != pending_tasks_.end(); ++i) {
       if (IsSequenceTokenRunnable(i->sequence_token_id)) {
         // Found a runnable task, mark the thread as being started.
         thread_being_created_ = true;
         return static_cast<int>(threads_.size() + 1);
       }
@@ skipping 158 matching lines @@
 bool SequencedWorkerPool::RunsTasksOnCurrentThread() const {
   return inner_->RunsTasksOnCurrentThread();
 }
 
 bool SequencedWorkerPool::IsRunningSequenceOnCurrentThread(
     SequenceToken sequence_token) const {
   return inner_->IsRunningSequenceOnCurrentThread(sequence_token);
 }
 
 void SequencedWorkerPool::FlushForTesting() {
-  inner_->FlushForTesting();
+  inner_->CleanupForTesting();
 }
 
 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