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();

[akalin, 2013/02/26 00:12:23]

any reason for this rename? perhaps make it consistent with the outer method
name (in a follow up CL perhaps)

 
   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) {

[akalin, 2013/02/26 00:12:23]

sad that cleanup handling became this complex, but if it's needed it's needed.
:-(

[michaeln, 2013/02/26 00:33:00]

I know... as mentioned a while ago... in some sense having this special cleanup
mode is an overall simplification (serialize cleanup and nuke it all), but the
details of entering and exiting that mode are not so simple.

+    // 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 62 matching lines @@
     if (!IsSequenceTokenRunnable(i->sequence_token_id)) {
       unrunnable_tasks++;
       ++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;
     }
 
     if (shutdown_called_ && i->shutdown_behavior != BLOCK_SHUTDOWN) {
       // We're shutting down and the task we just found isn't blocking
       // shutdown. Delete it and get more work.
       //
       // Note that we do not want to delete unrunnable tasks. Deleting a task
       // can have side effects (like freeing some objects) and deleting a
       // task that's supposed to run after one that's currently running could
@@ skipping 106 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