Fix |g_all_pools_state| data race by using atomics.

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 <stdint.h>
 
 #include <list>
 #include <map>
 #include <memory>
 #include <set>
 #include <utility>
 #include <vector>
 
 #include "base/atomic_sequence_num.h"
+#include "base/atomicops.h"
 #include "base/callback.h"
 #include "base/compiler_specific.h"
 #include "base/critical_closure.h"
 #include "base/lazy_instance.h"
 #include "base/logging.h"
 #include "base/macros.h"
 #include "base/memory/ptr_util.h"
 #include "base/stl_util.h"
 #include "base/strings/stringprintf.h"
 #include "base/synchronization/condition_variable.h"
@@ skipping 24 matching lines @@
 namespace base {
 
 namespace {
 
 // An enum representing the state of all pools. Any given process should only
 // ever transition from NONE_ACTIVE to the active states, transitions between
 // actives states are unexpected. The REDIRECTED_TO_TASK_SCHEDULER transition
 // occurs when RedirectSequencedWorkerPoolsToTaskSchedulerForProcess() is called
 // and the WORKER_CREATED transition occurs when a Worker needs to be created
 // because the first task was posted and the state is still NONE_ACTIVE.
+// |g_all_pools_state| uses relaxed atomic operations to ensure no data race
+// between reads/writes, strict memory ordering isn't required per no other
+// state being inferred from its value. Explicit synchronization (e.g. locks or
+// events) would be overkill (it's fine for other threads to still see
+// NONE_ACTIVE after the first Worker was created -- this is not possible for
+// REDIRECTED_TO_TASK_SCHEDULER per its API requesting to be invoked while no
+// other threads are active).
 // TODO(gab): Remove this if http://crbug.com/622400 fails (SequencedWorkerPool
 // will be phased out completely otherwise).
-enum class AllPoolsState {
+enum AllPoolsState : subtle::Atomic32 {
   NONE_ACTIVE,
   WORKER_CREATED,
   REDIRECTED_TO_TASK_SCHEDULER,
-} g_all_pools_state = AllPoolsState::NONE_ACTIVE;
+};
+subtle::Atomic32 g_all_pools_state = AllPoolsState::NONE_ACTIVE;
 
 struct SequencedTask : public TrackingInfo  {
   SequencedTask()
       : sequence_token_id(0),
         trace_id(0),
         sequence_task_number(0),
         shutdown_behavior(SequencedWorkerPool::BLOCK_SHUTDOWN) {}
 
   explicit SequencedTask(const tracked_objects::Location& from_here)
       : base::TrackingInfo(from_here, TimeTicks()),
@@ skipping 467 matching lines @@
 // Worker definitions ---------------------------------------------------------
 
 SequencedWorkerPool::Worker::Worker(
     scoped_refptr<SequencedWorkerPool> worker_pool,
     int thread_number,
     const std::string& prefix)
     : SimpleThread(prefix + StringPrintf("Worker%d", thread_number)),
       worker_pool_(std::move(worker_pool)),
       task_shutdown_behavior_(BLOCK_SHUTDOWN),
       is_processing_task_(false) {
-  DCHECK_NE(AllPoolsState::REDIRECTED_TO_TASK_SCHEDULER, g_all_pools_state);
+  DCHECK_EQ(AllPoolsState::WORKER_CREATED,
+            subtle::NoBarrier_Load(&g_all_pools_state));
   Start();
 }
 
 SequencedWorkerPool::Worker::~Worker() {
 }
 
 void SequencedWorkerPool::Worker::Run() {
-  DCHECK_NE(AllPoolsState::REDIRECTED_TO_TASK_SCHEDULER, g_all_pools_state);
+  DCHECK_EQ(AllPoolsState::WORKER_CREATED,
+            subtle::NoBarrier_Load(&g_all_pools_state));
 
 #if defined(OS_WIN)
   win::ScopedCOMInitializer com_initializer;
 #endif
 
   // Store a pointer to this worker in thread local storage for static function
   // access.
   DCHECK(!lazy_tls_ptr_.Get().Get());
   lazy_tls_ptr_.Get().Set(this);
 
@@ skipping 125 matching lines @@
         "SequencedWorkerPool::Inner::PostTask",
         TRACE_ID_MANGLE(GetTaskTraceID(sequenced, static_cast<void*>(this))),
         TRACE_EVENT_FLAG_FLOW_OUT);
 
     sequenced.sequence_task_number = LockedGetNextSequenceTaskNumber();
 
     // Now that we have the lock, apply the named token rules.
     if (optional_token_name)
       sequenced.sequence_token_id = LockedGetNamedTokenID(*optional_token_name);
 
-    if (g_all_pools_state == AllPoolsState::REDIRECTED_TO_TASK_SCHEDULER) {
+    if (subtle::NoBarrier_Load(&g_all_pools_state) ==
+        AllPoolsState::REDIRECTED_TO_TASK_SCHEDULER) {
       PostTaskToTaskScheduler(sequenced);
     } else {
       pending_tasks_.insert(sequenced);
 
       if (sequenced.shutdown_behavior == BLOCK_SHUTDOWN)
         blocking_shutdown_pending_task_count_++;
 
       create_thread_id = PrepareToStartAdditionalThreadIfHelpful();
     }
   }
 
-  if (g_all_pools_state != AllPoolsState::REDIRECTED_TO_TASK_SCHEDULER) {
+  if (subtle::NoBarrier_Load(&g_all_pools_state) !=
+      AllPoolsState::REDIRECTED_TO_TASK_SCHEDULER) {
     // Actually start the additional thread or signal an existing one outside
     // the lock.
     if (create_thread_id)
       FinishStartingAdditionalThread(create_thread_id);
     else
       SignalHasWork();
   }
 
 #if DCHECK_IS_ON()
   {
     AutoLock lock_for_dcheck(lock_);
     // Some variables are exposed in both modes for convenience but only really
     // intended for one of them at runtime, confirm exclusive usage here.
-    if (g_all_pools_state == AllPoolsState::REDIRECTED_TO_TASK_SCHEDULER) {
+    if (subtle::NoBarrier_Load(&g_all_pools_state) ==
+        AllPoolsState::REDIRECTED_TO_TASK_SCHEDULER) {
       DCHECK(pending_tasks_.empty());
       DCHECK_EQ(0, create_thread_id);
     } else {
       DCHECK(sequenced_task_runner_map_.empty());
     }
   }
 #endif  // DCHECK_IS_ON()
 
   return true;
 }
 
 void SequencedWorkerPool::Inner::PostTaskToTaskScheduler(
     const SequencedTask& sequenced) {
-  DCHECK_EQ(AllPoolsState::REDIRECTED_TO_TASK_SCHEDULER, g_all_pools_state);
+  DCHECK_EQ(AllPoolsState::REDIRECTED_TO_TASK_SCHEDULER,
+            subtle::NoBarrier_Load(&g_all_pools_state));
 
   lock_.AssertAcquired();
 
   // Confirm that the TaskScheduler's shutdown behaviors use the same
   // underlying values as SequencedWorkerPool.
   static_assert(
       static_cast<int>(TaskShutdownBehavior::CONTINUE_ON_SHUTDOWN) ==
           static_cast<int>(CONTINUE_ON_SHUTDOWN),
       "TaskShutdownBehavior and WorkerShutdown enum mismatch for "
       "CONTINUE_ON_SHUTDOWN.");
@@ skipping 41 matching lines @@
     // DCHECK ensures no such pools use SequencedWorkerPool::PostTask()
     // directly.
     DCHECK_GT(max_threads_, 1U);
     base::PostTaskWithTraits(sequenced.posted_from, pool_traits,
                              sequenced.task);
   }
 }
 
 bool SequencedWorkerPool::Inner::RunsTasksOnCurrentThread() const {
   AutoLock lock(lock_);
-  if (g_all_pools_state == AllPoolsState::REDIRECTED_TO_TASK_SCHEDULER) {
+  if (subtle::NoBarrier_Load(&g_all_pools_state) ==
+      AllPoolsState::REDIRECTED_TO_TASK_SCHEDULER) {
     if (!runs_tasks_on_verifier_) {
       runs_tasks_on_verifier_ = CreateTaskRunnerWithTraits(
           TaskTraits().WithFileIO().WithPriority(task_priority_),
           ExecutionMode::PARALLEL);
     }
     return runs_tasks_on_verifier_->RunsTasksOnCurrentThread();
   } else {
     return ContainsKey(threads_, PlatformThread::CurrentId());
   }
 }
 
 bool SequencedWorkerPool::Inner::IsRunningSequenceOnCurrentThread(
     SequenceToken sequence_token) const {
   AutoLock lock(lock_);
-  if (g_all_pools_state == AllPoolsState::REDIRECTED_TO_TASK_SCHEDULER) {
+  if (subtle::NoBarrier_Load(&g_all_pools_state) ==
+      AllPoolsState::REDIRECTED_TO_TASK_SCHEDULER) {
     // TODO(gab): This currently only verifies that the current thread is a
     // thread on which a task bound to |sequence_token| *could* run, but it
     // doesn't verify that the current is *currently running* a task bound to
     // |sequence_token|.
     const auto sequenced_task_runner_it =
         sequenced_task_runner_map_.find(sequence_token.id_);
     return sequenced_task_runner_it != sequenced_task_runner_map_.end() &&
            sequenced_task_runner_it->second->RunsTasksOnCurrentThread();
   } else {
     ThreadMap::const_iterator found =
@@ skipping 30 matching lines @@
     int max_new_blocking_tasks_after_shutdown) {
   DCHECK_GE(max_new_blocking_tasks_after_shutdown, 0);
   {
     AutoLock lock(lock_);
     // Cleanup and Shutdown should not be called concurrently.
     CHECK_EQ(CLEANUP_DONE, cleanup_state_);
     if (shutdown_called_)
       return;
     shutdown_called_ = true;
 
-    if (g_all_pools_state != AllPoolsState::WORKER_CREATED)
+    if (subtle::NoBarrier_Load(&g_all_pools_state) !=
+        AllPoolsState::WORKER_CREATED)
       return;
 
     max_blocking_tasks_after_shutdown_ = max_new_blocking_tasks_after_shutdown;
 
     // 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.
     SignalHasWork();
 
     // There are no pending or running tasks blocking shutdown, we're done.
     if (CanShutdown())
@@ skipping 21 matching lines @@
                       TimeTicks::Now() - shutdown_wait_begin);
 #endif
 }
 
 bool SequencedWorkerPool::Inner::IsShutdownInProgress() {
     AutoLock lock(lock_);
     return shutdown_called_;
 }
 
 void SequencedWorkerPool::Inner::ThreadLoop(Worker* this_worker) {
-  DCHECK_NE(AllPoolsState::REDIRECTED_TO_TASK_SCHEDULER, g_all_pools_state);
+  DCHECK_EQ(AllPoolsState::WORKER_CREATED,
+            subtle::NoBarrier_Load(&g_all_pools_state));
   {
     AutoLock lock(lock_);
     DCHECK(thread_being_created_);
     thread_being_created_ = false;
     auto result = threads_.insert(
         std::make_pair(this_worker->tid(), WrapUnique(this_worker)));
     DCHECK(result.second);
 
     while (true) {
 #if defined(OS_MACOSX)
@@ skipping 114 matching lines @@
 
   // 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).
   SignalHasWork();
 
   // Possibly unblock shutdown.
   can_shutdown_cv_.Signal();
 }
 
 void SequencedWorkerPool::Inner::HandleCleanup() {
-  DCHECK_EQ(AllPoolsState::WORKER_CREATED, g_all_pools_state);
+  DCHECK_EQ(AllPoolsState::WORKER_CREATED,
+            subtle::NoBarrier_Load(&g_all_pools_state));
 
   lock_.AssertAcquired();
   if (cleanup_state_ == CLEANUP_DONE)
     return;
   if (cleanup_state_ == CLEANUP_REQUESTED) {
     // We win, we get to do the cleanup as soon as the others wise up and idle.
     cleanup_state_ = CLEANUP_STARTING;
     while (thread_being_created_ ||
            cleanup_idlers_ != threads_.size() - 1) {
       has_work_cv_.Signal();
@@ skipping 46 matching lines @@
 int64_t SequencedWorkerPool::Inner::LockedGetNextSequenceTaskNumber() {
   lock_.AssertAcquired();
   // We assume that we never create enough tasks to wrap around.
   return next_sequence_task_number_++;
 }
 
 SequencedWorkerPool::Inner::GetWorkStatus SequencedWorkerPool::Inner::GetWork(
     SequencedTask* task,
     TimeDelta* wait_time,
     std::vector<Closure>* delete_these_outside_lock) {
-  DCHECK_EQ(AllPoolsState::WORKER_CREATED, g_all_pools_state);
+  DCHECK_EQ(AllPoolsState::WORKER_CREATED,
+            subtle::NoBarrier_Load(&g_all_pools_state));
 
   lock_.AssertAcquired();
 
   // 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
@@ skipping 67 matching lines @@
     }
 
     status = GET_WORK_FOUND;
     break;
   }
 
   return status;
 }
 
 int SequencedWorkerPool::Inner::WillRunWorkerTask(const SequencedTask& task) {
-  DCHECK_EQ(AllPoolsState::WORKER_CREATED, g_all_pools_state);
+  DCHECK_EQ(AllPoolsState::WORKER_CREATED,
+            subtle::NoBarrier_Load(&g_all_pools_state));
 
   lock_.AssertAcquired();
 
   // Mark the task's sequence number as in use.
   if (task.sequence_token_id)
     current_sequences_.insert(task.sequence_token_id);
 
   // Ensure that threads running tasks posted with either SKIP_ON_SHUTDOWN
   // or BLOCK_SHUTDOWN will prevent shutdown until that task or thread
   // completes.
@@ skipping 12 matching lines @@
   // if there is one waiting to pick up the next task.
   //
   // Note that we really need to do this *before* running the task, not
   // after. Otherwise, if more than one task is posted, the creation of the
   // 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) {
-  DCHECK_EQ(AllPoolsState::WORKER_CREATED, g_all_pools_state);
+  DCHECK_EQ(AllPoolsState::WORKER_CREATED,
+            subtle::NoBarrier_Load(&g_all_pools_state));
 
   lock_.AssertAcquired();
 
   if (task.shutdown_behavior != CONTINUE_ON_SHUTDOWN) {
     DCHECK_GT(blocking_shutdown_thread_count_, 0u);
     blocking_shutdown_thread_count_--;
   }
 
   if (task.sequence_token_id)
     current_sequences_.erase(task.sequence_token_id);
 }
 
 bool SequencedWorkerPool::Inner::IsSequenceTokenRunnable(
     int sequence_token_id) const {
-  DCHECK_NE(AllPoolsState::REDIRECTED_TO_TASK_SCHEDULER, g_all_pools_state);
+  DCHECK_NE(AllPoolsState::REDIRECTED_TO_TASK_SCHEDULER,
+            subtle::NoBarrier_Load(&g_all_pools_state));
 
   lock_.AssertAcquired();
   return !sequence_token_id ||
       current_sequences_.find(sequence_token_id) ==
           current_sequences_.end();
 }
 
 int SequencedWorkerPool::Inner::PrepareToStartAdditionalThreadIfHelpful() {
-  DCHECK_NE(AllPoolsState::REDIRECTED_TO_TASK_SCHEDULER, g_all_pools_state);
+  DCHECK_NE(AllPoolsState::REDIRECTED_TO_TASK_SCHEDULER,
+            subtle::NoBarrier_Load(&g_all_pools_state));
 
   lock_.AssertAcquired();
   // How thread creation works:
   //
   // We'de like to avoid creating threads with the lock held. However, we
   // need to be sure that we have an accurate accounting of the threads for
   // proper Joining and deltion on shutdown.
   //
   // We need to figure out if we need another thread with the lock held, which
   // is what this function does. It then marks us as in the process of creating
@@ skipping 31 matching lines @@
         thread_being_created_ = true;
         return static_cast<int>(threads_.size() + 1);
       }
     }
   }
   return 0;
 }
 
 void SequencedWorkerPool::Inner::FinishStartingAdditionalThread(
     int thread_number) {
-  DCHECK_NE(AllPoolsState::REDIRECTED_TO_TASK_SCHEDULER, g_all_pools_state);
+  DCHECK_NE(AllPoolsState::REDIRECTED_TO_TASK_SCHEDULER,
+            subtle::NoBarrier_Load(&g_all_pools_state));
 
   // Called outside of the lock.
   DCHECK_GT(thread_number, 0);
 
-  if (g_all_pools_state != AllPoolsState::WORKER_CREATED) {
-    DCHECK_EQ(AllPoolsState::NONE_ACTIVE, g_all_pools_state);
-    g_all_pools_state = AllPoolsState::WORKER_CREATED;
+  if (subtle::NoBarrier_Load(&g_all_pools_state) !=
+      AllPoolsState::WORKER_CREATED) {
+    DCHECK_EQ(AllPoolsState::NONE_ACTIVE,
+              subtle::NoBarrier_Load(&g_all_pools_state));
+    subtle::NoBarrier_Store(&g_all_pools_state, AllPoolsState::WORKER_CREATED);
   }
 
   // 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() {
-  DCHECK_NE(AllPoolsState::REDIRECTED_TO_TASK_SCHEDULER, g_all_pools_state);
+  DCHECK_NE(AllPoolsState::REDIRECTED_TO_TASK_SCHEDULER,
+            subtle::NoBarrier_Load(&g_all_pools_state));
 
   has_work_cv_.Signal();
   if (testing_observer_) {
     testing_observer_->OnHasWork();
   }
 }
 
 bool SequencedWorkerPool::Inner::CanShutdown() const {
-  DCHECK_EQ(AllPoolsState::WORKER_CREATED, g_all_pools_state);
+  DCHECK_EQ(AllPoolsState::WORKER_CREATED,
+            subtle::NoBarrier_Load(&g_all_pools_state));
   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_;
 
@@ skipping 24 matching lines @@
 }
 
 // static
 void SequencedWorkerPool::
     RedirectSequencedWorkerPoolsToTaskSchedulerForProcess() {
   DCHECK(TaskScheduler::GetInstance());
   // Hitting this DCHECK indicates that a task was posted to a
   // SequencedWorkerPool before the TaskScheduler was initialized and
   // redirected, posting task to SequencedWorkerPools needs to at least be
   // delayed until after that point.
-  DCHECK_EQ(AllPoolsState::NONE_ACTIVE, g_all_pools_state);
-  g_all_pools_state = AllPoolsState::REDIRECTED_TO_TASK_SCHEDULER;
+  DCHECK_EQ(AllPoolsState::NONE_ACTIVE,
+            subtle::NoBarrier_Load(&g_all_pools_state));
+  subtle::NoBarrier_Store(&g_all_pools_state,
+                          AllPoolsState::REDIRECTED_TO_TASK_SCHEDULER);
 }
 
 SequencedWorkerPool::SequencedWorkerPool(size_t max_threads,
                                          const std::string& thread_name_prefix,
                                          base::TaskPriority task_priority)
     : constructor_task_runner_(ThreadTaskRunnerHandle::Get()),
       inner_(new Inner(this,
                        max_threads,
                        thread_name_prefix,
                        task_priority,
@@ skipping 138 matching lines @@
 void SequencedWorkerPool::Shutdown(int max_new_blocking_tasks_after_shutdown) {
   DCHECK(constructor_task_runner_->BelongsToCurrentThread());
   inner_->Shutdown(max_new_blocking_tasks_after_shutdown);
 }
 
 bool SequencedWorkerPool::IsShutdownInProgress() {
   return inner_->IsShutdownInProgress();
 }
 
 }  // namespace base