Hook up the SequencedWorkerPool to the browser thread.

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 <deque>
 #include <set>
 
 #include "base/atomicops.h"
@@ skipping 39 matching lines @@
 
 
 // Inner ----------------------------------------------------------------------
 
 class SequencedWorkerPool::Inner
     : public base::RefCountedThreadSafe<SequencedWorkerPool::Inner> {
  public:
   Inner(size_t max_threads, const std::string& thread_name_prefix);
   virtual ~Inner();
 
-  // Backends for SequenceWorkerPool.
   SequenceToken GetSequenceToken();
+
   SequenceToken GetNamedSequenceToken(const std::string& name);
-  bool PostTask(int sequence_token_id,
+
+  // This function accepts a name and an ID. If the name is null, the
+  // token ID is used. This allows us to implement the optional name lookup
+  // from a single function without having to enter the lock a separate time.
+  bool PostTask(const std::string* optional_token_name,
+                int sequence_token_id,
                 SequencedWorkerPool::WorkerShutdown shutdown_behavior,
                 const tracked_objects::Location& from_here,
                 const base::Closure& task);
+
+  void Flush();
+
   void Shutdown();
+
   void SetTestingObserver(SequencedWorkerPool::TestingObserver* observer);
 
   // Runs the worker loop on the background thread.
   void ThreadLoop(Worker* this_worker);
 
  private:
+  // 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<base::Closure>* delete_these_outside_lock);
 
   // 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.
@@ skipping 144 matching lines @@
 SequencedWorkerPool::Inner::GetSequenceToken() {
   base::subtle::Atomic32 result =
       base::subtle::NoBarrier_AtomicIncrement(&last_sequence_number_, 1);
   return SequenceToken(static_cast<int>(result));
 }
 
 SequencedWorkerPool::SequenceToken
 SequencedWorkerPool::Inner::GetNamedSequenceToken(
     const std::string& name) {
   base::AutoLock lock(lock_);
-  std::map<std::string, int>::const_iterator found =
-      named_sequence_tokens_.find(name);
-  if (found != named_sequence_tokens_.end())
-    return SequenceToken(found->second);  // Got an existing one.
-
-  // Create a new one for this name.
-  SequenceToken result = GetSequenceToken();
-  named_sequence_tokens_.insert(std::make_pair(name, result.id_));
-  return result;
+  return SequenceToken(LockedGetNamedTokenID(name));
 }
 
 bool SequencedWorkerPool::Inner::PostTask(
+    const std::string* optional_token_name,
     int sequence_token_id,
     SequencedWorkerPool::WorkerShutdown shutdown_behavior,
     const tracked_objects::Location& from_here,
     const base::Closure& task) {
   SequencedTask sequenced;
   sequenced.sequence_token_id = sequence_token_id;
   sequenced.shutdown_behavior = shutdown_behavior;
   sequenced.location = from_here;
   sequenced.task = task;
 
   int create_thread_id = 0;
   {
     base::AutoLock lock(lock_);
     if (terminating_)
       return false;
 
+    // Now that we have the lock, apply the named token rules.
+    if (optional_token_name)
+      sequenced.sequence_token_id = LockedGetNamedTokenID(*optional_token_name);
+
     pending_tasks_.push_back(sequenced);
     pending_task_count_++;
     if (shutdown_behavior == BLOCK_SHUTDOWN)
       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();
 
   return true;
 }
 
+void SequencedWorkerPool::Inner::Flush() {
+  {
+    base::AutoLock lock(lock_);
+    while (pending_task_count_ > 0 || waiting_thread_count_ < threads_.size())
+      cond_var_.Wait();
+  }
+  cond_var_.Signal();
+}
+
 void SequencedWorkerPool::Inner::Shutdown() {
   if (shutdown_called_)
     return;
   shutdown_called_ = true;
 
   // 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.
   {
@@ skipping 65 matching lines @@
         }
         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 terminating_ 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 (terminating_)
           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_--;
       }
     }
   }
 
   // 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();
 }
 
+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())
+    return found->second;  // Got an existing one.
+
+  // Create a new one for this name.
+  SequenceToken result = GetSequenceToken();
+  named_sequence_tokens_.insert(std::make_pair(name, result.id_));
+  return result.id_;
+}
+
 bool SequencedWorkerPool::Inner::GetWork(
     SequencedTask* task,
     std::vector<base::Closure>* delete_these_outside_lock) {
   lock_.AssertAcquired();
 
   DCHECK_EQ(pending_tasks_.size(), pending_task_count_);
   UMA_HISTOGRAM_COUNTS_100("SequencedWorkerPool.TaskCount",
                            static_cast<int>(pending_task_count_));
 
   // Find the next task with a sequence token that's not currently in use.
@@ skipping 196 matching lines @@
 }
 
 SequencedWorkerPool::SequenceToken SequencedWorkerPool::GetNamedSequenceToken(
     const std::string& name) {
   return inner_->GetNamedSequenceToken(name);
 }
 
 bool SequencedWorkerPool::PostWorkerTask(
     const tracked_objects::Location& from_here,
     const base::Closure& task) {
-  return inner_->PostTask(0, BLOCK_SHUTDOWN, from_here, task);
+  return inner_->PostTask(NULL, 0, BLOCK_SHUTDOWN, from_here, task);
 }
 
 bool SequencedWorkerPool::PostWorkerTaskWithShutdownBehavior(
     const tracked_objects::Location& from_here,
     const base::Closure& task,
     WorkerShutdown shutdown_behavior) {
-  return inner_->PostTask(0, shutdown_behavior, from_here, task);
+  return inner_->PostTask(NULL, 0, shutdown_behavior, from_here, task);
 }
 
 bool SequencedWorkerPool::PostSequencedWorkerTask(
     SequenceToken sequence_token,
     const tracked_objects::Location& from_here,
     const base::Closure& task) {
-  return inner_->PostTask(sequence_token.id_, BLOCK_SHUTDOWN,
+  return inner_->PostTask(NULL, sequence_token.id_, BLOCK_SHUTDOWN,
                           from_here, task);
 }
 
+bool SequencedWorkerPool::PostNamedSequencedWorkerTask(
+    const std::string& token_name,
+    const tracked_objects::Location& from_here,
+    const base::Closure& task) {
+  DCHECK(!token_name.empty());
+  return inner_->PostTask(&token_name, 0, BLOCK_SHUTDOWN, from_here, task);
+}
+
 bool SequencedWorkerPool::PostSequencedWorkerTaskWithShutdownBehavior(
     SequenceToken sequence_token,
     const tracked_objects::Location& from_here,
     const base::Closure& task,
     WorkerShutdown shutdown_behavior) {
-  return inner_->PostTask(sequence_token.id_, shutdown_behavior,
+  return inner_->PostTask(NULL, sequence_token.id_, shutdown_behavior,
                           from_here, task);
 }
 
+void SequencedWorkerPool::FlushForTesting() {
+  inner_->Flush();
+}
+
 void SequencedWorkerPool::Shutdown() {
   inner_->Shutdown();
 }
 
 void SequencedWorkerPool::SetTestingObserver(TestingObserver* observer) {
   inner_->SetTestingObserver(observer);
 }
 
 }  // namespace base