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,
   };
 
+  // Helper used by PostTask() to complete the work when redirection is on.
+  // Coalesce upon resolution of http://crbug.com/622400.
+  void PostTaskToTaskScheduler(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 28 matching lines @@
 
   // Checks if all threads are busy and the addition of one more could run an
   // additional task waiting in the queue. This must be called from within
   // the lock.
   //
   // If another thread is helpful, this will mark the thread as being in the
   // process of starting and returns the index of the new thread which will be
   // 0 or more. The caller should then call FinishStartingAdditionalThread to
   // complete initialization once the lock is released.
   //
-  // If another thread is not necessary, returne 0;
+  // If another thread is not necessary, return 0;
   //
   // See the implementedion for more.
   int PrepareToStartAdditionalThreadIfHelpful();
 
   // The second part of thread creation after
   // PrepareToStartAdditionalThreadIfHelpful with the thread number it
   // generated. This actually creates the thread and should be called outside
   // the lock to avoid blocking important work starting a thread in the lock.
   void FinishStartingAdditionalThread(int thread_number);
 
@@ skipping 79 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);
+
 #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 88 matching lines @@
   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_);
+

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

(you added whitespace here as a result)

[gab, 2016/09/07 15:00:10]

True, but I kind of think it's cleaner that way as the statement applies to the
whole block not just the following one?

     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 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() &&
@@ skipping 15 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);
 
-    pending_tasks_.insert(sequenced);
-    if (shutdown_behavior == BLOCK_SHUTDOWN)
-      blocking_shutdown_pending_task_count_++;
+    if (g_all_pools_state == AllPoolsState::REDIRECTED_TO_TASK_SCHEDULER) {
+      PostTaskToTaskScheduler(sequenced);
+    } else {
+      pending_tasks_.insert(sequenced);
 
-    create_thread_id = PrepareToStartAdditionalThreadIfHelpful();
+      if (sequenced.shutdown_behavior == BLOCK_SHUTDOWN)
+        blocking_shutdown_pending_task_count_++;
+
+      create_thread_id = PrepareToStartAdditionalThreadIfHelpful();
+    }
   }
 
-  // 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();
+  if (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) {
+      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) {

[fdoray, 2016/08/25 15:14:01]

This method doesn't honor delays :(

[gab, 2016/09/07 15:00:10]

Oops, looks like you're addressing that in
https://codereview.chromium.org/2285633003/? Thanks!

+  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);

[fdoray, 2016/08/25 15:14:01]

SequencedWorkerPool allows tasks with different shutdown behaviors to be posted
to the same sequence. TaskScheduler currently doesn't support that. Do we want
to DCHECK that this doesn't happen for tasks forwarded to the TaskScheduler?

[gab, 2016/09/07 15:00:10]

Yes, let's do that (seems like that's what you're doing in
https://codereview.chromium.org/2285633003/ :-))

+
+  // 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;

[fdoray, 2016/08/25 15:14:01]

Should we DCHECK_LE(sequenced_task_runner_map_.size(), 1)? Otherwise, tasks
posted to different sequences on the same single-threaded SequencedWorkerPool
may run on different threads.

[gab, 2016/09/07 15:00:10]

Good point, I think this is the case in practice so that sounds fine to me.

+      *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 currently
+    // legitimatelly assume thread affinity despite using 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);
+  }
+}
+
 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);

[fdoray, 2016/08/25 15:14:01]

Should we have a special case for single-threaded SequencedWorkerPools here?

[gab, 2016/09/07 15:00:10]

Yes, looks like you got this too :-)

+    }
+    return runs_tasks_on_verifier_->RunsTasksOnCurrentThread();
+  } else {
+    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 {
+    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);
   {
     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 136 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