Add SequencedWorkerPool.IsRunningSequenceOnCurrentThread so callers can make stronger assertions ab…

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/lazy_instance.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/thread_local.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 {
@@ skipping 18 matching lines @@
   // Hold a (cyclic) ref to |worker_pool|, since we want to keep it
   // around as long as we are running.
   Worker(const scoped_refptr<SequencedWorkerPool>& worker_pool,
          int thread_number,
          const std::string& thread_name_prefix);
   virtual ~Worker();
 
   // SimpleThread implementation. This actually runs the background thread.
   virtual void Run() OVERRIDE;
 
+
+  void set_running_sequence(SequenceToken token) {
+    DCHECK_EQ(current(), this);
+    running_sequence_ = token;
+  }
+
+  bool is_running_sequence(SequenceToken token) const {
+    DCHECK_EQ(current(), this);
+    return running_sequence_.Equals(token);
+  }
+
+  SequencedWorkerPool* owner() const { 
+    DCHECK_EQ(current(), this);
+    return worker_pool_.get();
+  }
+
+  static Worker* current() {
+    return current_worker_tls_.Pointer()->Get();
+  }
+
  private:
   scoped_refptr<SequencedWorkerPool> worker_pool_;
+  SequenceToken running_sequence_;
+  static LazyInstance<ThreadLocalPointer<Worker> > current_worker_tls_;
 
   DISALLOW_COPY_AND_ASSIGN(Worker);
 };
 
+LazyInstance<ThreadLocalPointer<SequencedWorkerPool::Worker> >
+    SequencedWorkerPool::Worker::current_worker_tls_ =
+        LAZY_INSTANCE_INITIALIZER;
+
 // Inner ----------------------------------------------------------------------
 
 class SequencedWorkerPool::Inner {
  public:
   // Take a raw pointer to |worker| to avoid cycles (since we're owned
   // by it).
   Inner(SequencedWorkerPool* worker_pool, size_t max_threads,
         const std::string& thread_name_prefix,
         TestingObserver* observer);
 
   ~Inner();
 
   SequenceToken GetSequenceToken();
 
   SequenceToken GetNamedSequenceToken(const std::string& name);
 
   // 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;
 
+  bool IsRunningSequenceOnCurrentThread(SequenceToken sequence_token) const;
+
   void FlushForTesting();
 
   void SignalHasWorkForTesting();
 
   int GetWorkSignalCountForTesting() const;
 
   void Shutdown();
 
   // Runs the worker loop on the background thread.
   void ThreadLoop(Worker* this_worker);
@@ skipping 138 matching lines @@
 
 SequencedWorkerPool::Worker::~Worker() {
 }
 
 void SequencedWorkerPool::Worker::Run() {
   // Just jump back to the Inner object to run the thread, since it has all the
   // tracking information and queues. It might be more natural to implement
   // using DelegateSimpleThread and have Inner implement the Delegate to avoid
   // having these worker objects at all, but that method lacks the ability to
   // send thread-specific information easily to the thread loop.
+  current_worker_tls_.Pointer()->Set(this);
   worker_pool_->inner_->ThreadLoop(this);
   // Release our cyclic reference once we're done.
   worker_pool_ = NULL;
+  // Reset TLS after the worker_pool_ is reference is dropped because
+  // its destructor tests Worker::current().
+  current_worker_tls_.Pointer()->Set(NULL);
 }
 
 // Inner definitions ---------------------------------------------------------
 
 SequencedWorkerPool::Inner::Inner(
     SequencedWorkerPool* worker_pool,
     size_t max_threads,
     const std::string& thread_name_prefix,
     TestingObserver* observer)
     : worker_pool_(worker_pool),
@@ skipping 72 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
     SignalHasWork();
 
   return true;
 }
 
 bool SequencedWorkerPool::Inner::RunsTasksOnCurrentThread() const {

[michaeln, 2012/03/25 00:20:02]

we could hoist these method bodies into SequencedWorkerPool directly and remove
these two Inner methods

-  AutoLock lock(lock_);
-  return ContainsKey(threads_, PlatformThread::CurrentId());
+  Worker* worker = Worker::current();
+  return worker && (worker->owner() == worker_pool_);
+}
+
+bool SequencedWorkerPool::Inner::IsRunningSequenceOnCurrentThread(
+    SequenceToken sequence_token) const {
+  Worker* worker = Worker::current();
+  return worker && (worker->owner() == worker_pool_) &&
+         worker->is_running_sequence(sequence_token);
 }
 
 void SequencedWorkerPool::Inner::FlushForTesting() {
   AutoLock lock(lock_);
   while (!IsIdle())
     is_idle_cv_.Wait();
 }
 
 void SequencedWorkerPool::Inner::SignalHasWorkForTesting() {
   SignalHasWork();
@@ skipping 59 matching lines @@
           // worker thread. (Technically not required, since we
           // already get a signal for each new task, but it doesn't
           // hurt.)
           SignalHasWork();
           delete_these_outside_lock.clear();
 
           // Complete thread creation outside the lock if necessary.
           if (new_thread_id)
             FinishStartingAdditionalThread(new_thread_id);
 
+          this_worker->set_running_sequence(
+              SequenceToken(task.sequence_token_id));
+
           task.task.Run();
 
+          this_worker->set_running_sequence(SequenceToken());
+
           // 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
@@ skipping 260 matching lines @@
       inner_(new Inner(ALLOW_THIS_IN_INITIALIZER_LIST(this),
                        max_threads, thread_name_prefix, observer)) {
 }
 
 SequencedWorkerPool::~SequencedWorkerPool() {}
 
 void SequencedWorkerPool::OnDestruct() const {
   DCHECK(constructor_message_loop_.get());
   // Avoid deleting ourselves on a worker thread (which would
   // deadlock).
-  if (RunsTasksOnCurrentThread()) {
+  if (Worker::current()) {
     constructor_message_loop_->DeleteSoon(FROM_HERE, this);
   } else {
     delete this;
   }
 }
 
 SequencedWorkerPool::SequenceToken SequencedWorkerPool::GetSequenceToken() {
   return inner_->GetSequenceToken();
 }
 
@@ skipping 58 matching lines @@
     TimeDelta delay) {
   // TODO(akalin): Add support for non-zero delays.
   DCHECK_EQ(delay.InMillisecondsRoundedUp(), 0);
   return PostWorkerTask(from_here, task);
 }
 
 bool SequencedWorkerPool::RunsTasksOnCurrentThread() const {
   return inner_->RunsTasksOnCurrentThread();
 }
 
+bool SequencedWorkerPool::IsRunningSequenceOnCurrentThread(
+    SequenceToken sequence_token) const {
+  return inner_->IsRunningSequenceOnCurrentThread(sequence_token);
+}
+
 void SequencedWorkerPool::FlushForTesting() {
   inner_->FlushForTesting();
 }
 
 void SequencedWorkerPool::SignalHasWorkForTesting() {
   inner_->SignalHasWorkForTesting();
 }
 
 void SequencedWorkerPool::Shutdown() {
   DCHECK(constructor_message_loop_->BelongsToCurrentThread());
   inner_->Shutdown();
 }
 
 }  // namespace base