Clean up condition variable usage in SequencedWorkerPool

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 75 matching lines @@
   bool PostTask(const std::string* optional_token_name,
                 SequenceToken sequence_token,
                 WorkerShutdown shutdown_behavior,
                 const tracked_objects::Location& from_here,
                 const Closure& task);
 
   bool RunsTasksOnCurrentThread() const;
 
   void FlushForTesting();
 
+  void TriggerSpuriousWorkSignalForTesting();
+
+  int GetWorkSignalCountForTesting() const;
+
   void Shutdown();
 
   void SetTestingObserver(TestingObserver* observer);
 
   // Runs the worker loop on the background thread.
   void ThreadLoop(Worker* this_worker);
 
  private:
+  // Returns whether there are no more pending tasks and all threads
+  // are idle.  Must be called under lock.
+  bool IsIdle() const;
+
   // 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);
 
   // The calling code should clear the given delete_these_oustide_lock
   // vector the next time the lock is released. See the implementation for
   // a more detailed description.
   bool GetWork(SequencedTask* task,
                std::vector<Closure>* delete_these_outside_lock);
 
@@ skipping 21 matching lines @@
   //
   // See the implementedion for more.
   int PrepareToStartAdditionalThreadIfHelpful();
 
   // The second part of thread creation after
   // PrepareToStartAdditionalThreadIfHelpful with the thread number it
   // generated. This actually creates the thread and should be called outside
   // the lock to avoid blocking important work starting a thread in the lock.
   void FinishStartingAdditionalThread(int thread_number);
 
+  // Signal |has_work_| and increment |has_work_signal_count_|.
+  void SignalHasWork();
+
   // Checks whether there is work left that's blocking shutdown. Must be
   // called inside the lock.
   bool CanShutdown() const;
 
   SequencedWorkerPool* const worker_pool_;
 
   // The last sequence number used. Managed by GetSequenceToken, since this
   // only does threadsafe increment operations, you do not need to hold the
   // lock.
   volatile subtle::Atomic32 last_sequence_number_;
 
   // 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 used to wake up worker threads when a task is runnable.
-  ConditionVariable cond_var_;
+  // Condition variable that is signalled whenever a new task is
+  // posted or when Shutdown() is called.
+  ConditionVariable has_work_cv_;
+
+  // Number of times |has_work_| has been signalled.  Used for testing.
+  int has_work_signal_count_;
+
+  // Condition variable that is signalled whenever IsIdle() goes to
+  // true.
+  ConditionVariable is_idle_cv_;
+
+  // Condition variable that is signalled whwnever 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_;
 
   // Owning pointers to all threads we've created so far, indexed by
@@ skipping 65 matching lines @@
 
 // Inner definitions ---------------------------------------------------------
 
 SequencedWorkerPool::Inner::Inner(
     SequencedWorkerPool* worker_pool,
     size_t max_threads,
     const std::string& thread_name_prefix)
     : worker_pool_(worker_pool),
       last_sequence_number_(0),
       lock_(),
-      cond_var_(&lock_),
+      has_work_cv_(&lock_),
+      has_work_signal_count_(0),
+      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),
       pending_task_count_(0),
       blocking_shutdown_pending_task_count_(0),
       shutdown_called_(false),
       testing_observer_(NULL)  {}
 
@@ skipping 52 matching lines @@
       blocking_shutdown_pending_task_count_++;
 
     create_thread_id = PrepareToStartAdditionalThreadIfHelpful();
   }
 
   // Actually start the additional thread or signal an existing one now that
   // we're outside the lock.
   if (create_thread_id)
     FinishStartingAdditionalThread(create_thread_id);
   else
-    cond_var_.Signal();
+    SignalHasWork();
 
   return true;
 }
 
 bool SequencedWorkerPool::Inner::RunsTasksOnCurrentThread() const {
   AutoLock lock(lock_);
   return ContainsKey(threads_, PlatformThread::CurrentId());
 }
 
 void SequencedWorkerPool::Inner::FlushForTesting() {
-  {
-    AutoLock lock(lock_);
-    while (pending_task_count_ > 0 || waiting_thread_count_ < threads_.size())
-      cond_var_.Wait();
-  }
-  cond_var_.Signal();
+  AutoLock lock(lock_);
+  while (!IsIdle())
+    is_idle_cv_.Wait();
+}
+
+void SequencedWorkerPool::Inner::TriggerSpuriousWorkSignalForTesting() {
+  SignalHasWork();
+}
+
+int SequencedWorkerPool::Inner::GetWorkSignalCountForTesting() const {

[jar (doing other things), 2012/03/12 19:42:36]

nit: It bugs me a little that we've added this must testing infrastructure...
and the optimizer will probably never be able to realize these are not needed. 
I wish these could be done in the unit test file, with minimal access given to a
FriendTest class.  For instance, it would be nice if we only handed the test
class private access to our lock, or such, and let it craft CVs to its heart
content :-/.

For now, I'll focus on this CL, and its correctness... but I wanted to pass on
that thought.

[akalin, 2012/03/12 20:42:16]

I think I can remove this variable and instead add another method to the
TestingObserver class that is called whenever has_work_cv_ is signalled.  I'll
do that in another CL.

+  AutoLock lock(lock_);
+  return has_work_signal_count_;
 }
 
 void SequencedWorkerPool::Inner::Shutdown() {
   // Mark us as terminated and go through and drop all tasks that aren't
   // required to run on shutdown. Since no new tasks will get posted once the
   // terminated flag is set, this ensures that all remaining tasks are required
   // for shutdown whenever the termianted_ flag is set.
   {
     AutoLock lock(lock_);
 
     if (shutdown_called_)
       return;
     shutdown_called_ = true;
 
     // 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.
-    cond_var_.Signal();
+    has_work_cv_.Signal();
 
     // There are no pending or running tasks blocking shutdown, we're done.
     if (CanShutdown())
       return;
   }
 
-  // If we get here, we know we're either waiting on a blocking task that's
-  // currently running, waiting on a blocking task that hasn't been scheduled
-  // yet, or both. Block on the "queue empty" event to know when all tasks are
-  // complete. This must be done outside the lock.
+  // 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();
 
-  // Wait for no more tasks.
   {
     AutoLock lock(lock_);
     while (!CanShutdown())
-      cond_var_.Wait();
+      can_shutdown_cv_.Wait();
   }
   UMA_HISTOGRAM_TIMES("SequencedWorkerPool.ShutdownDelayTime",
                       TimeTicks::Now() - shutdown_wait_begin);
 }
 
 void SequencedWorkerPool::Inner::SetTestingObserver(
     TestingObserver* observer) {
   AutoLock lock(lock_);
   testing_observer_ = observer;
 }
 
 void SequencedWorkerPool::Inner::ThreadLoop(Worker* this_worker) {
   {
     AutoLock lock(lock_);
     DCHECK(thread_being_created_);
     thread_being_created_ = false;
     std::pair<ThreadMap::iterator, bool> result =
         threads_.insert(
             std::make_pair(this_worker->tid(), make_linked_ptr(this_worker)));
     DCHECK(result.second);
 
     while (true) {
       // See GetWork for what delete_these_outside_lock is doing.
       SequencedTask task;
       std::vector<Closure> delete_these_outside_lock;
       if (GetWork(&task, &delete_these_outside_lock)) {
         int new_thread_id = WillRunWorkerTask(task);
         {
           AutoUnlock unlock(lock_);
-          cond_var_.Signal();
+          // There may be more work available, so wake up another
+          // worker thread.
+          has_work_cv_.Signal();
           delete_these_outside_lock.clear();
 
           // Complete thread creation outside the lock if necessary.
           if (new_thread_id)
             FinishStartingAdditionalThread(new_thread_id);
 
           task.task.Run();
 
           // Make sure our task is erased outside the lock for the same reason
           // we do this with delete_these_oustide_lock.
           task.task = Closure();
         }
         DidRunWorkerTask(task);  // Must be done inside the lock.
       } else {
         // When we're terminating and there's no more work, we can
         // shut down.  You can't get more tasks posted once
         // shutdown_called_ is set. 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_)
           break;
         waiting_thread_count_++;
-        cond_var_.Signal();  // For Flush() that may be waiting on the
-                             // waiting thread count to go up.
-        cond_var_.Wait();
+        // This is the only time that IsIdle() can go to true.
+        if (IsIdle())
+          is_idle_cv_.Signal();
+        has_work_cv_.Wait();
         waiting_thread_count_--;
       }
     }
+
+    // Unblock shutdown if possible.
+    if (CanShutdown())
+      can_shutdown_cv_.Signal();

[jar (doing other things), 2012/03/12 19:42:36]

Placing this here is probably more optimal than I had envisioned.  I figured
you'd just put the extra signal outside the lock, and let the controlling
process (if any) check to see if CanShutdown() (which it is already definitely
planning to check, watching for spurious signals!).

Please add the comment for the closing brace below, so It will be clear that
this flow only happens when we are about to exit (and we've broken out of the
loop).

[akalin, 2012/03/12 20:42:16]

Yeah, I think that's simpler.  I moved it out.

   }

[jar (doing other things), 2012/03/12 19:42:36]

nit: Add comment: // Release lock_.

[akalin, 2012/03/12 20:42:16]

Done.

 
   // 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).
-  cond_var_.Signal();
+  has_work_cv_.Signal();
+}
+
+bool SequencedWorkerPool::Inner::IsIdle() const {
+  lock_.AssertAcquired();
+  return pending_task_count_ == 0 && waiting_thread_count_ == threads_.size();
 }
 
 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 63 matching lines @@
       // vector they passed to us once the lock is exited to make this
       // happen.
       delete_these_outside_lock->push_back(i->task);
       i = pending_tasks_.erase(i);
       pending_task_count_--;
     } else {
       // Found a runnable task.
       *task = *i;
       i = pending_tasks_.erase(i);
       pending_task_count_--;
-      if (task->shutdown_behavior == BLOCK_SHUTDOWN)
+      if (task->shutdown_behavior == BLOCK_SHUTDOWN) {
         blocking_shutdown_pending_task_count_--;
+      }
 
       found_task = true;
       break;
     }
   }
 
   // Track the number of tasks we had to skip over to see if we should be
   // making this more efficient. If this number ever becomes large or is
   // frequently "some", we should consider the optimization above.
   UMA_HISTOGRAM_COUNTS_100("SequencedWorkerPool.UnrunnableTaskCount",
@@ skipping 100 matching lines @@
 void SequencedWorkerPool::Inner::FinishStartingAdditionalThread(
     int thread_number) {
   // Called outside of the lock.
   DCHECK(thread_number > 0);
 
   // The worker is assigned to the list when the thread actually starts, which
   // will manage the memory of the pointer.
   new Worker(worker_pool_, thread_number, thread_name_prefix_);
 }
 
+void SequencedWorkerPool::Inner::SignalHasWork() {
+  has_work_cv_.Signal();
+  {
+    AutoLock lock(lock_);
+    ++has_work_signal_count_;
+  }
+}
+
 bool SequencedWorkerPool::Inner::CanShutdown() const {
   lock_.AssertAcquired();
   // See PrepareToStartAdditionalThreadIfHelpful for how thread creation works.
   return !thread_being_created_ &&
          blocking_shutdown_thread_count_ == 0 &&
          blocking_shutdown_pending_task_count_ == 0;
 }
 
 // SequencedWorkerPool --------------------------------------------------------
 
 SequencedWorkerPool::SequencedWorkerPool(
     size_t max_threads,
     const std::string& thread_name_prefix)
     : constructor_message_loop_(MessageLoopProxy::current()),
       inner_(new Inner(ALLOW_THIS_IN_INITIALIZER_LIST(this),
                        max_threads, thread_name_prefix)) {
   DCHECK(constructor_message_loop_.get());
 }
 
 SequencedWorkerPool::~SequencedWorkerPool() {}
 
 void SequencedWorkerPool::OnDestruct() const {
   // TODO(akalin): Once we can easily check if we're on a worker
   // thread or not, use that instead of restricting destruction to
   // only the constructor message loop.
-  if (constructor_message_loop_->BelongsToCurrentThread()) {
+  if (constructor_message_loop_->BelongsToCurrentThread())
     delete this;
-  } else {
+  else
     constructor_message_loop_->DeleteSoon(FROM_HERE, this);
-  }
 }
 
 SequencedWorkerPool::SequenceToken SequencedWorkerPool::GetSequenceToken() {
   return inner_->GetSequenceToken();
 }
 
 SequencedWorkerPool::SequenceToken SequencedWorkerPool::GetNamedSequenceToken(
     const std::string& name) {
   return inner_->GetNamedSequenceToken(name);
 }
@@ skipping 58 matching lines @@
 }
 
 bool SequencedWorkerPool::RunsTasksOnCurrentThread() const {
   return inner_->RunsTasksOnCurrentThread();
 }
 
 void SequencedWorkerPool::FlushForTesting() {
   inner_->FlushForTesting();
 }
 
+void SequencedWorkerPool::TriggerSpuriousWorkSignalForTesting() {
+  inner_->TriggerSpuriousWorkSignalForTesting();
+}
+
+int SequencedWorkerPool::GetWorkSignalCountForTesting() const {
+  return inner_->GetWorkSignalCountForTesting();
+}
+
 void SequencedWorkerPool::Shutdown() {
+  DCHECK(constructor_message_loop_->BelongsToCurrentThread());
   inner_->Shutdown();
 }
 
 void SequencedWorkerPool::SetTestingObserver(TestingObserver* observer) {
   inner_->SetTestingObserver(observer);
 }
 
 }  // namespace base