Implement SequencedWorkerPool::RunsTasksOnCurrentThread()

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>
 #include <vector>
 
 #include "base/atomicops.h"
 #include "base/callback.h"
 #include "base/compiler_specific.h"
 #include "base/logging.h"
 #include "base/memory/linked_ptr.h"
 #include "base/message_loop_proxy.h"
 #include "base/metrics/histogram.h"
+#include "base/stl_util.h"
 #include "base/stringprintf.h"
 #include "base/synchronization/condition_variable.h"
 #include "base/synchronization/lock.h"
+#include "base/threading/platform_thread.h"
 #include "base/threading/simple_thread.h"
 #include "base/time.h"
 #include "base/tracked_objects.h"
 
 namespace base {
 
 namespace {
 
 struct SequencedTask {
   SequencedTask()
@@ skipping 47 matching lines @@
 
   // 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,
                 SequenceToken sequence_token,
                 WorkerShutdown shutdown_behavior,
                 const tracked_objects::Location& from_here,
                 const Closure& task);
 
+  bool RunsTasksOnCurrentThread() const;
+
   void FlushForTesting();
 
   void Shutdown();
 
   void SetTestingObserver(TestingObserver* observer);
 
   // Runs the worker loop on the background thread.
   void ThreadLoop(Worker* this_worker);
 
  private:
@@ skipping 45 matching lines @@
   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.
-  Lock lock_;
+  mutable Lock lock_;
 
   // Condition variable used to wake up worker threads when a task is runnable.
   ConditionVariable cond_var_;
 
   // 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. Since we lazily
-  // create threads, this may be less than max_threads_ and will be initially
-  // empty.
-  std::vector<linked_ptr<Worker> > threads_;
+  // Owning pointers to all threads we've created so far, indexed by
+  // ID. Since we lazily create threads, this may be less than
+  // max_threads_ and will be initially empty.
+  typedef std::map<PlatformThreadId, linked_ptr<Worker> > ThreadMap;
+  ThreadMap threads_;
 
   // Set to true when we're in the process of creating another thread.
   // See PrepareToStartAdditionalThreadIfHelpful for more.
   bool thread_being_created_;
 
   // Number of threads currently waiting for work.
   size_t waiting_thread_count_;
 
   // Number of threads currently running tasks that have the BLOCK_SHUTDOWN
   // flag set.
@@ skipping 68 matching lines @@
       blocking_shutdown_pending_task_count_(0),
       shutdown_called_(false),
       testing_observer_(NULL)  {}
 
 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 (size_t i = 0; i < threads_.size(); i++)
-    threads_[i]->Join();
+  for (ThreadMap::iterator it = threads_.begin(); it != threads_.end(); ++it)
+    it->second->Join();
   threads_.clear();
 
   if (testing_observer_)
     testing_observer_->OnDestruct();
 }
 
 SequencedWorkerPool::SequenceToken
 SequencedWorkerPool::Inner::GetSequenceToken() {
   subtle::Atomic32 result =
       subtle::NoBarrier_AtomicIncrement(&last_sequence_number_, 1);
@@ skipping 39 matching lines @@
   // 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;
 }
 
+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();
 }
 
 void SequencedWorkerPool::Inner::Shutdown() {
@@ skipping 40 matching lines @@
     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;
-    threads_.push_back(linked_ptr<Worker>(this_worker));
+    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();
@@ skipping 331 matching lines @@
 bool SequencedWorkerPool::PostDelayedTask(
     const tracked_objects::Location& from_here,
     const Closure& task,
     TimeDelta delay) {
   // TODO(akalin): Add support for non-zero delays.
   DCHECK_EQ(delay.InMillisecondsRoundedUp(), 0);
   return PostWorkerTask(from_here, task);
 }
 
 bool SequencedWorkerPool::RunsTasksOnCurrentThread() const {
-  // TODO(akalin): Keep track of the thread IDs of our worker threads
-  // and use those to implement this function.
-  NOTREACHED();
-  return true;
+  return inner_->RunsTasksOnCurrentThread();
 }
 
 void SequencedWorkerPool::FlushForTesting() {
   inner_->FlushForTesting();
 }
 
 void SequencedWorkerPool::Shutdown() {
   inner_->Shutdown();
 }
 
 void SequencedWorkerPool::SetTestingObserver(TestingObserver* observer) {
   inner_->SetTestingObserver(observer);
 }
 
 }  // namespace base