Add an experiment to redirect SequencedWorkerPool tasks to TaskScheduler.

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/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"
 #include "base/synchronization/lock.h"
+#include "base/task_scheduler/post_task.h"
+#include "base/task_scheduler/task_scheduler.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/threading/thread_task_runner_handle.h"
 #include "base/time/time.h"
 #include "base/trace_event/heap_profiler.h"
 #include "base/trace_event/trace_event.h"
 #include "base/tracked_objects.h"
 #include "build/build_config.h"
 
 #if defined(OS_MACOSX)
 #include "base/mac/scoped_nsautorelease_pool.h"
 #elif defined(OS_WIN)
 #include "base/win/scoped_com_initializer.h"
 #endif
 
 #if !defined(OS_NACL)
 #include "base/metrics/histogram.h"
 #endif
 
 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.
+// TODO(gab): Remove this if http://crbug.com/622400 fails (SequencedWorkerPool
+// will be phased out completely otherwise).
+enum class AllPoolsState {
+  NONE_ACTIVE,
+  WORKER_CREATED,
+  REDIRECTED_TO_TASK_SCHEDULER,
+} 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()),
         sequence_token_id(0),
@@ skipping 284 matching lines @@
   };
 
   enum CleanupState {
     CLEANUP_REQUESTED,
     CLEANUP_STARTING,
     CLEANUP_RUNNING,
     CLEANUP_FINISHING,
     CLEANUP_DONE,
   };
 
+  // Helpers used by PostTask() to complete the work depending on whether
+  // redirection is on. Coalesce upon resolution of http://crbug.com/622400.
+  void PostTaskToTaskScheduler(const SequencedTask& sequenced);
+  void PostTaskToPool(const SequencedTask& sequenced);
+
   // 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);
 
   // Called from within the lock, this returns the next sequence task number.
   int64_t LockedGetNextSequenceTaskNumber();
 
   // Gets new task. There are 3 cases depending on the return value:
   //
   // 1) If the return value is |GET_WORK_FOUND|, |task| is filled in and should
@@ skipping 128 matching lines @@
   // has been called.
   int max_blocking_tasks_after_shutdown_;
 
   // State used to cleanup for testing, all guarded by lock_.
   CleanupState cleanup_state_;
   size_t cleanup_idlers_;
   ConditionVariable cleanup_cv_;
 
   TestingObserver* const testing_observer_;
 
+  // Members below are used for the experimental redirection to TaskScheduler.
+  // TODO(gab): Remove these if http://crbug.com/622400 fails
+  // (SequencedWorkerPool will be phased out completely otherwise).
+
   // The TaskPriority to be used for SequencedWorkerPool tasks redirected to the
   // TaskScheduler as an experiment (unused otherwise).
   const base::TaskPriority task_priority_;
 
+  // A map of SequenceToken IDs to TaskScheduler TaskRunners used to redirect
+  // SequencedWorkerPool usage to the TaskScheduler.
+  std::map<int, scoped_refptr<TaskRunner>> sequenced_task_runner_map_;
+
+  // A dummy TaskRunner obtained from TaskScheduler with the same TaskTraits as
+  // used by this SequencedWorkerPool to query for RunsTasksOnCurrentThread().
+  // Mutable so it can be lazily instantiated from RunsTasksOnCurrentThread().
+  mutable scoped_refptr<TaskRunner> runs_tasks_on_verifier_;
+
   DISALLOW_COPY_AND_ASSIGN(Inner);
 };
 
 // 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);
   Start();
 }
 
 SequencedWorkerPool::Worker::~Worker() {
 }
 
 void SequencedWorkerPool::Worker::Run() {
+  DCHECK_NE(AllPoolsState::REDIRECTED_TO_TASK_SCHEDULER, g_all_pools_state);

[danakj, 2016/08/15 21:02:52]

Run is happening before RedirectSequencedWorkerPoolsToTaskSchedulerForProcess?

[gab, 2016/08/15 21:32:44]

No, this documents that Run() should never happen in the redirected world.
g_all_pools_state is used to enforce we either transition to the redirected
state or the regular workers state but that we never call redirect after a
Worker was started. It is also used to document (like here) which methods are
exclusive to a given mode.

+
 #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);
 
   // Just jump back to the Inner object to run the thread, since it has all the
@@ skipping 85 matching lines @@
   DCHECK(delay.is_zero() || shutdown_behavior == SKIP_ON_SHUTDOWN);
   SequencedTask sequenced(from_here);
   sequenced.sequence_token_id = sequence_token.id_;
   sequenced.shutdown_behavior = shutdown_behavior;
   sequenced.posted_from = from_here;
   sequenced.task =
       shutdown_behavior == BLOCK_SHUTDOWN ?
       base::MakeCriticalClosure(task) : task;
   sequenced.time_to_run = TimeTicks::Now() + delay;
 
-  int create_thread_id = 0;
-  {
-    AutoLock lock(lock_);
-    if (shutdown_called_) {
-      // Don't allow a new task to be posted if it doesn't block shutdown.
-      if (shutdown_behavior != BLOCK_SHUTDOWN)
-        return false;
+  AutoLock lock(lock_);
 
-      // If the current thread is running a task, and that task doesn't block
-      // shutdown, then it shouldn't be allowed to post any more tasks.
-      ThreadMap::const_iterator found =
-          threads_.find(PlatformThread::CurrentId());
-      if (found != threads_.end() && found->second->is_processing_task() &&
-          found->second->task_shutdown_behavior() != BLOCK_SHUTDOWN) {
-        return false;
-      }
+  if (shutdown_called_) {
+    // Don't allow a new task to be posted if it doesn't block shutdown.
+    if (shutdown_behavior != BLOCK_SHUTDOWN)
+      return false;
 
-      if (max_blocking_tasks_after_shutdown_ <= 0) {
-        DLOG(WARNING) << "BLOCK_SHUTDOWN task disallowed";
-        return false;
-      }
-      max_blocking_tasks_after_shutdown_ -= 1;
+    // If the current thread is running a task, and that task doesn't block
+    // shutdown, then it shouldn't be allowed to post any more tasks.
+    ThreadMap::const_iterator found =
+        threads_.find(PlatformThread::CurrentId());
+    if (found != threads_.end() && found->second->is_processing_task() &&
+        found->second->task_shutdown_behavior() != BLOCK_SHUTDOWN) {
+      return false;
     }
 
-    // The trace_id is used for identifying the task in about:tracing.
-    sequenced.trace_id = trace_id_++;
-
-    TRACE_EVENT_WITH_FLOW0(TRACE_DISABLED_BY_DEFAULT("toplevel.flow"),
-        "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);
-
-    pending_tasks_.insert(sequenced);
-    if (shutdown_behavior == BLOCK_SHUTDOWN)
-      blocking_shutdown_pending_task_count_++;
-
-    create_thread_id = PrepareToStartAdditionalThreadIfHelpful();
+    if (max_blocking_tasks_after_shutdown_ <= 0) {
+      DLOG(WARNING) << "BLOCK_SHUTDOWN task disallowed";
+      return false;
+    }
+    max_blocking_tasks_after_shutdown_ -= 1;
   }
 
-  // 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);
+  // The trace_id is used for identifying the task in about:tracing.
+  sequenced.trace_id = trace_id_++;
+
+  TRACE_EVENT_WITH_FLOW0(TRACE_DISABLED_BY_DEFAULT("toplevel.flow"),
+      "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) {
+    PostTaskToTaskScheduler(sequenced);
+  } else {
+    PostTaskToPool(sequenced);
+  }
+
+  // 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)
+    DCHECK(pending_tasks_.empty());
   else
-    SignalHasWork();
+    DCHECK(sequenced_task_runner_map_.empty());
 
   return true;
 }
 
+void SequencedWorkerPool::Inner::PostTaskToTaskScheduler(
+    const SequencedTask& sequenced) {
+  DCHECK_EQ(AllPoolsState::REDIRECTED_TO_TASK_SCHEDULER, 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.");
+  static_assert(static_cast<int>(TaskShutdownBehavior::SKIP_ON_SHUTDOWN) ==
+                    static_cast<int>(SKIP_ON_SHUTDOWN),
+                "TaskShutdownBehavior and WorkerShutdown enum mismatch for "
+                "SKIP_ON_SHUTDOWN.");
+  static_assert(static_cast<int>(TaskShutdownBehavior::BLOCK_SHUTDOWN) ==
+                    static_cast<int>(BLOCK_SHUTDOWN),
+                "TaskShutdownBehavior and WorkerShutdown enum mismatch for "
+                "BLOCK_SHUTDOWN.");
+
+  const TaskShutdownBehavior task_shutdown_behavior =
+      static_cast<TaskShutdownBehavior>(sequenced.shutdown_behavior);
+  const TaskTraits pool_traits =
+      TaskTraits()
+          .WithFileIO()
+          .WithPriority(task_priority_)
+          .WithShutdownBehavior(task_shutdown_behavior);
+
+  // Find or create the TaskScheduler TaskRunner to redirect this task to if
+  // it is posted to a specific sequence.
+  scoped_refptr<TaskRunner>* sequenced_task_runner = nullptr;
+  if (sequenced.sequence_token_id) {
+    sequenced_task_runner =
+        &sequenced_task_runner_map_[sequenced.sequence_token_id];
+    if (!*sequenced_task_runner) {
+      const ExecutionMode execution_mode =
+          max_threads_ == 1U ? ExecutionMode::SINGLE_THREADED
+                             : ExecutionMode::SEQUENCED;
+      *sequenced_task_runner =
+          CreateTaskRunnerWithTraits(pool_traits, execution_mode);
+    }
+  }
+
+  if (sequenced_task_runner) {
+    (*sequenced_task_runner)
+        ->PostTask(sequenced.posted_from, sequenced.task);
+  } else {
+    // PostTaskWithTraits() posts a task with PARALLEL semantics. There are
+    // however a few  pools that use only one thread and therefore can

[danakj, 2016/08/15 21:02:52]

nit: "few  pools" extra space

[gab, 2016/08/15 21:32:44]

Done.

+    // currently legitimatelly assuming thread affinity despite using

[danakj, 2016/08/15 21:02:52]

"legitimately assume"

[gab, 2016/08/15 21:32:44]

Done.

+    // SequencedWorkerPool. Such pools typically only give access to their
+    // TaskRunner which will be SINGLE_THREADED per nature of the pool
+    // having only one thread but this DCHECK ensures no such pools use
+    // SequencedWorkerPool::PostTask() directly.
+    DCHECK_GT(max_threads_, 1U);
+    base::PostTaskWithTraits(sequenced.posted_from, pool_traits,
+                             sequenced.task);
+  }
+}
+
+void SequencedWorkerPool::Inner::PostTaskToPool(
+    const SequencedTask& sequenced) {
+  DCHECK_NE(AllPoolsState::REDIRECTED_TO_TASK_SCHEDULER, g_all_pools_state);
+
+  lock_.AssertAcquired();
+
+  pending_tasks_.insert(sequenced);
+
+  if (sequenced.shutdown_behavior == BLOCK_SHUTDOWN)
+    blocking_shutdown_pending_task_count_++;
+
+  int create_thread_id = PrepareToStartAdditionalThreadIfHelpful();
+
+  {
+    AutoUnlock unlock(lock_);

[brettw, 2016/08/15 21:02:21]

Sorry to add another pass here. I'm really reluctant to add another set of
locking for every task we post to the worker pool today. It seems like this is a
common operation now, nobody uses the new codepath yet, and lock contention is
bad.

The structure you have here is definitely more clear than the alternatives, but
I think it's worth compromising a bit to avoid the lock. Can we keep this in the
PostTask function like we did before (outside the lock) and only run it when
we're in the old mode?

[danakj, 2016/08/15 21:02:52]

hm, this will relock the lock when you leave the block tho, adding contention.
I'd prefer you put the AutoLock into something (a unique_ptr would work but
require malloc, a base::Optional would also work and not require malloc), and
move that thing into the Post method here so it can release it early.

[gab, 2016/08/15 21:32:44]

Re-inlined PostTaskToPool() -- as without this section it doesn't do much and
requires having it returning int and documenting that it's the same as
PrepareToStartAdditionalThreadIfHelpful() which just makes it cumbersome IMO.

[gab, 2016/08/15 21:32:44]

N/A after re-inlining PostTaskToPool per Brett's comment.

+
+    // Actually start the additional thread or signal an existing one outside
+    // the lock.
+    if (create_thread_id)
+      FinishStartingAdditionalThread(create_thread_id);
+    else
+      SignalHasWork();
+  }
+}
+
 bool SequencedWorkerPool::Inner::RunsTasksOnCurrentThread() const {
-  AutoLock lock(lock_);
-  return ContainsKey(threads_, PlatformThread::CurrentId());
+  if (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 {
+    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->is_processing_task() &&
-         sequence_token.Equals(found->second->task_sequence_token());
+  if (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 {
+    AutoLock lock(lock_);
+    ThreadMap::const_iterator found =
+        threads_.find(PlatformThread::CurrentId());
+    if (found == threads_.end())
+      return false;
+    return found->second->is_processing_task() &&
+           sequence_token.Equals(found->second->task_sequence_token());
+  }
 }
 
 // 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 13 matching lines @@
 void SequencedWorkerPool::Inner::Shutdown(
     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)
+      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())
       return;
   }
@@ skipping 19 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);

[danakj, 2016/08/15 21:02:52]

Why are some of these != REDIRECTED instead of == POOL? We'd make threads before
setting the mode?

[gab, 2016/08/15 21:32:44]

We never explicitly set POOL (a.k.a. WORKER_CREATED) from a static call.
WORKER_CREATED is set when a worker is created. For that mode it's valid for the
exclusive methods to run either before the mode is determined or after, it
doesn't matter, as long as the redirection was explicitly asked for.

It's true that in this case on ThreadLoop() it should always already be
determined but I opted to document as == REDIRECTED and != REDIRECTED everywhere
which is really the split I'm trying to highlight.

[danakj, 2016/08/15 22:05:49]

Hm ok. I would personally prefer == checks whenever it's possible since it gives
the reader more information. This is a bit ambiguous it could mean it can happen
earlier too. But you're also disappearing and I don't feel like strongly enough
to block this or anything.

[gab, 2016/08/20 15:21:31]

Unless the first Worker is created while only the main thread is around,
checking for == WORKER_CREATED is racy from pre-existing threads (this is why
the API for the redirection is strict on it being called before any other
threads are created). This is why no logic is based on it being set and also why
I think it's more correct to not use it in DCHECKs either (although == would be
fine in logic that only runs from the worker threads themselves -- I can do that
when I'm back if you strongly prefer).

   {
     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);
+
   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();
       cleanup_cv_.Wait();
@@ skipping 45 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);
+
   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
   // there might be work to schedule. If this proves to be a problem, we
@@ skipping 66 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);
+
   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.
   if (task.shutdown_behavior != CONTINUE_ON_SHUTDOWN)
@@ skipping 11 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);
+
   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);
+
   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);
+
   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
   // a thread. When we do shutdown, we wait until the thread_being_created_
@@ skipping 30 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);
+
   // 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;
+  }
+
   // 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);
+
   has_work_cv_.Signal();
   if (testing_observer_) {
     testing_observer_->OnHasWork();
   }
 }
 
 bool SequencedWorkerPool::Inner::CanShutdown() const {
+  DCHECK_EQ(AllPoolsState::WORKER_CREATED, 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 16 matching lines @@
 // static
 scoped_refptr<SequencedWorkerPool>
 SequencedWorkerPool::GetWorkerPoolForCurrentThread() {
   Worker* worker = Worker::GetForCurrentThread();
   if (!worker)
     return nullptr;
 
   return worker->worker_pool();
 }
 
+// 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;
+}
+
 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,
                        NULL)) {}
 
@@ skipping 142 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