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 {
+  AutoLock lock(lock_);
+  return has_work_signal_count_;
 }
 
 void SequencedWorkerPool::Inner::Shutdown() {

[jar (doing other things), 2012/03/12 17:39:13]

IMO, it would be helpful if you commented, or asserted, that this is NOT run on
a worker thread.  I think it is probably run on only one thread... and it would
be nicest if we could assert we're on said thread.

[akalin, 2012/03/12 18:34:34]

Done. (Added comment, and assert is in SWP::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();
+    // Wake up all waiting threads.
+    has_work_cv_.Broadcast();

[jar (doing other things), 2012/03/12 17:39:13]

I liked the original code better, relying on signalling, rather than
Broadcast(), which is plagued by the "box car" problem (threads wake up only to
block on a lock).  The good news is this is happening only at shutdown... so it
may not be very perf critical.

If you have distinct CVs (as are currently proposed in this CL), one for
signalling wake-ups, and one for signalling shutdown progress, then you should
IMO leave the code as it was (using Signal()), as it is cleaner and efficient.

If you'd like to re-use the CV (stop trying to add additional condition
variables), then you could use Broadcast() as suggested.  That would wake all
sleeping workers. They would then each either terminate, or run tasks (and
eventually terminate), as we are in a shutdown mode.  This would mean that the
CV Signal()ing (hence forth) would only be targetted at this controlling thread.

If you decide to keep the Broadcast(), and reuse the singular CV, then this is a
pretty subtle transition IMO. There should be a comment to the effect that
worker threads will no longer ever Wait() on this.

[akalin, 2012/03/12 18:34:34]

Okay, moved back to signal chain.

 
     // 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

[jar (doing other things), 2012/03/12 17:39:13]

This comment doesn't relate to any code IMO.
In addition, queue empty doesn't mean that there is nothing left blocking us....
so it is very confused.

[akalin, 2012/03/12 18:34:34]

Replaced comment (and the one below) with something clearer (IMO).

   // complete. This must be done outside the lock.
   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) {
-  {

[jar (doing other things), 2012/03/12 17:39:13]

nit: You are dead right that this curly is not needed (if/after you make your
change deleting line 435)... but it is making it much harder to see what you
changed in this function (blame it on the diff system).  Given how delicate this
CL is, I'd rather leave it in for now, so that I can better see the changes
here.  Thanks in advance.

[akalin, 2012/03/12 18:34:34]

Done. (It ended up being needed anyway when addressing your comments.)

-    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);
+  AutoLock lock(lock_);
+  DCHECK(thread_being_created_);
+  // This can make CanShutdown() go to true.
+  thread_being_created_ = false;
+  if (CanShutdown())

[jar (doing other things), 2012/03/12 17:39:13]

I don't think anyone cares about CanShutdown() unless we have shutdown_called_.
I expect the login in the running loop to check this status, and so I don't
expect to need to see special cases on thread startup here.

Bottom line: Why are lines 422 and 423 needed?  In this class of code, less is
better IMO.

[akalin, 2012/03/12 18:34:34]

Done.  Moved can_shutdown_cv_ signal to end of block.

+    can_shutdown_cv_.Signal();
+  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();

[jar (doing other things), 2012/03/12 17:39:13]

This signal is basically asking that anyone else that is waiting should check to
see if they have work to do.  Why did you delete it here?

[akalin, 2012/03/12 18:34:34]

Hmm.  If every task posted causes a signal, then wouldn't another thread be
woken up anyway?  But put it back in nonetheless.

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

I *think* you are correct.  This is a slightly (IMO) abnormal pattern that we
get from this use of CVs, in that we get a signal for each new task, and work
doesn't just spontaneously start appearing without signals (i.e., tasks in our
queue don't just quietly insert more tasks).  

IMO, it is a little subtle, and I'm used to a simple "always signal when you
take on work and can't Wait()," pattern.

I think this might therefore be overkill... but ends up having no real
consequence (I think).  It sure makes it easy to see correctness though, IMO.

On 2012/03/12 18:34:34, akalin wrote:

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

I see.  Okay, I just added a comment as to why it's left in.

-          delete_these_outside_lock.clear();
+  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_);
+        delete_these_outside_lock.clear();
 
-          // Complete thread creation outside the lock if necessary.
-          if (new_thread_id)
-            FinishStartingAdditionalThread(new_thread_id);
+        // Complete thread creation outside the lock if necessary.
+        if (new_thread_id)
+          FinishStartingAdditionalThread(new_thread_id);
 
-          task.task.Run();
+        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();
-        waiting_thread_count_--;
+        // 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_++;
+      // This is the only time that IsIdle() can go to true.
+      if (IsIdle())
+        is_idle_cv_.Signal();
+      has_work_cv_.Wait();
+      waiting_thread_count_--;
     }
   }
+}
 
-  // 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();

[jar (doing other things), 2012/03/12 17:39:13]

This is another fairly critical Signal() call. This indicates that we've decided
to stop running tasks, and are terminating.  We notify other waiting threads in
case they too wish to follow suit.

Why was this deleted?

At best, playing with your multiple-cv plan, I'd expect the addition of:
can_shutdown_cv_.Signal();
Surly when we are terminating, there is a *chance* that shutdown can and should
proceed.... as we might have been the thread folks were waiting for.

[akalin, 2012/03/12 18:34:34]

This was deleted because of the change to using Broadcast().  Now that I
reverted that, I put this back in.
Removed all the other places I signal can_shutdown_cv_ and added it here.

+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) {
+        // This can make CanShutdown() go to true.
         blocking_shutdown_pending_task_count_--;
+        if (CanShutdown())
+          can_shutdown_cv_.Signal();

[jar (doing other things), 2012/03/12 17:39:13]

Again, I don't like seeing special case code interspersed and signalling this
CV.  I think the right place to signal this is when a thread does indeed
shutdown, because it can find nothing more to do.

[akalin, 2012/03/12 18:34:34]

Done (see above)

+      }
 
       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 27 matching lines @@
   // second thread (since we only create one at a time) will be blocked by
   // the execution of the first task, which could be arbitrarily long.
   return PrepareToStartAdditionalThreadIfHelpful();
 }
 
 void SequencedWorkerPool::Inner::DidRunWorkerTask(const SequencedTask& task) {
   lock_.AssertAcquired();
 
   if (task.shutdown_behavior == BLOCK_SHUTDOWN) {
     DCHECK_GT(blocking_shutdown_thread_count_, 0u);
+    // This can make CanShutdown() go to true.
     blocking_shutdown_thread_count_--;
+    if (CanShutdown())
+      can_shutdown_cv_.Signal();

[jar (doing other things), 2012/03/12 17:39:13]

Again, please don't add special case code for this.  Try to put it in one
central and most logical place (see comments above).

[akalin, 2012/03/12 18:34:34]

Done (see above)

   }
 
   if (task.sequence_token_id)
     current_sequences_.erase(task.sequence_token_id);
 }
 
 bool SequencedWorkerPool::Inner::IsSequenceTokenRunnable(
     int sequence_token_id) const {
   lock_.AssertAcquired();
   return !sequence_token_id ||
@@ skipping 52 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();
+  ++has_work_signal_count_;

[jar (doing other things), 2012/03/12 17:39:13]

You are not generally inside a lock, but are manipulating this counter which
does require a lock.

I suspect that this testing variable should not be used... but I'm not sure.

[akalin, 2012/03/12 18:34:34]

My mistake.  This testing variable is kind of hacky, but it's needed for the
regression test for the bug this is supposed to fix.

+}
+
 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() {
   inner_->Shutdown();
 }
 
 void SequencedWorkerPool::SetTestingObserver(TestingObserver* observer) {
   inner_->SetTestingObserver(observer);
 }
 
 }  // namespace base