base: Make SequencedWorkerPool issue globally unique SequenceTokens.

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/atomic_sequence_num.h"
 #include "base/callback.h"
 #include "base/compiler_specific.h"
 #include "base/critical_closure.h"
 #include "base/debug/trace_event.h"
+#include "base/lazy_instance.h"
 #include "base/logging.h"
 #include "base/memory/linked_ptr.h"
 #include "base/message_loop/message_loop_proxy.h"
-#include "base/metrics/histogram.h"
 #include "base/stl_util.h"
 #include "base/strings/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/threading/thread_local.h"
 #include "base/threading/thread_restrictions.h"
 #include "base/time/time.h"
 #include "base/tracked_objects.h"
 
 #if defined(OS_MACOSX)
 #include "base/mac/scoped_nsautorelease_pool.h"
 #endif
 
+#if !defined(OS_NACL)
+#include "base/metrics/histogram.h"
+#endif
+
 namespace base {
 
 namespace {
 
 struct SequencedTask : public TrackingInfo  {
   SequencedTask()
       : sequence_token_id(0),
         trace_id(0),
         sequence_task_number(0),
         shutdown_behavior(SequencedWorkerPool::BLOCK_SHUTDOWN) {}
@@ skipping 160 matching lines @@
 
 // Create a process-wide unique ID to represent this task in trace events. This
 // will be mangled with a Process ID hash to reduce the likelyhood of colliding
 // with MessageLoop pointers on other processes.
 uint64 GetTaskTraceID(const SequencedTask& task,
                       void* pool) {
   return (static_cast<uint64>(task.trace_id) << 32) |
          static_cast<uint64>(reinterpret_cast<intptr_t>(pool));
 }
 
+base::LazyInstance<base::ThreadLocalPointer<
+    SequencedWorkerPool::SequenceToken> > g_lazy_tls_ptr =
+        LAZY_INSTANCE_INITIALIZER;
+
 }  // namespace
 
 // Worker ---------------------------------------------------------------------
 
 class SequencedWorkerPool::Worker : public SimpleThread {
  public:
   // 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,
@@ skipping 149 matching lines @@
   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_;
+  // lock. This is class-static to make SequenceTokens issued by
+  // GetSequenceToken unique across SequencedWorkerPool instances.
+  static base::StaticAtomicSequenceNumber g_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 that is waited on by worker threads until new
   // tasks are posted or shutdown starts.
   ConditionVariable has_work_cv_;
 
@@ skipping 75 matching lines @@
           prefix + StringPrintf("Worker%d", thread_number).c_str()),
       worker_pool_(worker_pool),
       running_shutdown_behavior_(CONTINUE_ON_SHUTDOWN) {
   Start();
 }
 
 SequencedWorkerPool::Worker::~Worker() {
 }
 
 void SequencedWorkerPool::Worker::Run() {
+  // Store a pointer to the running sequence in thread local storage for
+  // static function access.
+  g_lazy_tls_ptr.Get().Set(&running_sequence_);
+
   // 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.
   worker_pool_->inner_->ThreadLoop(this);
   // Release our cyclic reference once we're done.
   worker_pool_ = 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),
-      last_sequence_number_(0),
       lock_(),
       has_work_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),
       next_sequence_task_number_(0),
       blocking_shutdown_pending_task_count_(0),
@@ skipping 14 matching lines @@
   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);
-  return SequenceToken(static_cast<int>(result));
+  // Need to add one because StaticAtomicSequenceNumber starts at zero, which
+  // is used as a sentinel value in SequenceTokens.
+  return SequenceToken(g_last_sequence_number_.GetNext() + 1);
 }
 
 SequencedWorkerPool::SequenceToken
 SequencedWorkerPool::Inner::GetNamedSequenceToken(const std::string& name) {
   AutoLock lock(lock_);
   return SequenceToken(LockedGetNamedTokenID(name));
 }
 
 bool SequencedWorkerPool::Inner::PostTask(
     const std::string* optional_token_name,
@@ skipping 60 matching lines @@
   AutoLock lock(lock_);
   return ContainsKey(threads_, PlatformThread::CurrentId());
 }
 
 bool SequencedWorkerPool::Inner::IsRunningSequenceOnCurrentThread(
     SequenceToken sequence_token) const {
   AutoLock lock(lock_);
   ThreadMap::const_iterator found = threads_.find(PlatformThread::CurrentId());
   if (found == threads_.end())
     return false;
-  return found->second->running_sequence().Equals(sequence_token);
+  return sequence_token.Equals(found->second->running_sequence());
 }
 
 // See https://code.google.com/p/chromium/issues/detail?id=168415
 void SequencedWorkerPool::Inner::CleanupForTesting() {
   DCHECK(!RunsTasksOnCurrentThread());
   base::ThreadRestrictions::ScopedAllowWait allow_wait;
   AutoLock lock(lock_);
   CHECK_EQ(CLEANUP_DONE, cleanup_state_);
   if (shutdown_called_)
     return;
@@ skipping 30 matching lines @@
     if (CanShutdown())
       return;
   }
 
   // If we're here, then something is blocking shutdown.  So wait for
   // CanShutdown() to go to true.
 
   if (testing_observer_)
     testing_observer_->WillWaitForShutdown();
 
+#if !defined(OS_NACL)
   TimeTicks shutdown_wait_begin = TimeTicks::Now();
+#endif
 
   {
     base::ThreadRestrictions::ScopedAllowWait allow_wait;
     AutoLock lock(lock_);
     while (!CanShutdown())
       can_shutdown_cv_.Wait();
   }
+#if !defined(OS_NACL)
   UMA_HISTOGRAM_TIMES("SequencedWorkerPool.ShutdownDelayTime",
                       TimeTicks::Now() - shutdown_wait_begin);
+#endif
 }
 
 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)));
@@ skipping 176 matching lines @@
     return CONTINUE_ON_SHUTDOWN;
   return found->second->running_shutdown_behavior();
 }
 
 SequencedWorkerPool::Inner::GetWorkStatus SequencedWorkerPool::Inner::GetWork(
     SequencedTask* task,
     TimeDelta* wait_time,
     std::vector<Closure>* delete_these_outside_lock) {
   lock_.AssertAcquired();
 
+#if !defined(OS_NACL)
   UMA_HISTOGRAM_COUNTS_100("SequencedWorkerPool.TaskCount",
                            static_cast<int>(pending_tasks_.size()));
+#endif
 
   // Find the next task with a sequence token that's not currently in use.
   // If the token is in use, that means another thread is running something
   // in that sequence, and we can't run it without going out-of-order.
   //
   // This algorithm is simple and fair, but inefficient in some cases. For
   // example, say somebody schedules 1000 slow tasks with the same sequence
   // number. We'll have to go through all those tasks each time we feel like
   // there might be work to schedule. If this proves to be a problem, we
   // should make this more efficient.
@@ skipping 64 matching lines @@
       blocking_shutdown_pending_task_count_--;
     }
 
     status = GET_WORK_FOUND;
     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.
+#if !defined(OS_NACL)
   UMA_HISTOGRAM_COUNTS_100("SequencedWorkerPool.UnrunnableTaskCount",
                            unrunnable_tasks);
+#endif
   return status;
 }
 
 int SequencedWorkerPool::Inner::WillRunWorkerTask(const SequencedTask& task) {
   lock_.AssertAcquired();
 
   // Mark the task's sequence number as in use.
   if (task.sequence_token_id)
     current_sequences_.insert(task.sequence_token_id);
 
@@ skipping 108 matching lines @@
 }
 
 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;
 }
 
+base::StaticAtomicSequenceNumber
+SequencedWorkerPool::Inner::g_last_sequence_number_;
+
 // SequencedWorkerPool --------------------------------------------------------
 
+// static
+SequencedWorkerPool::SequenceToken
+SequencedWorkerPool::GetSequenceTokenForCurrentThread() {
+  SequencedWorkerPool::SequenceToken* token = g_lazy_tls_ptr.Get().Get();
+  if (!token)
+    return SequenceToken();
+  return *token;
+}
+
 SequencedWorkerPool::SequencedWorkerPool(
     size_t max_threads,
     const std::string& thread_name_prefix)
     : constructor_message_loop_(MessageLoopProxy::current()),
       inner_(new Inner(this, max_threads, thread_name_prefix, NULL)) {
 }
 
 SequencedWorkerPool::SequencedWorkerPool(
     size_t max_threads,
     const std::string& thread_name_prefix,
@@ skipping 127 matching lines @@
 void SequencedWorkerPool::SignalHasWorkForTesting() {
   inner_->SignalHasWorkForTesting();
 }
 
 void SequencedWorkerPool::Shutdown(int max_new_blocking_tasks_after_shutdown) {
   DCHECK(constructor_message_loop_->BelongsToCurrentThread());
   inner_->Shutdown(max_new_blocking_tasks_after_shutdown);
 }
 
 }  // namespace base