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.

[brettw, 2016/08/12 19:30:00]

This should have a reference to a bug to remove it.

[gab, 2016/08/12 23:27:45]

Done.

+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 432 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 used for the experimental redirection to TaskScheduler. */

[brettw, 2016/08/12 17:44:39]

Can this be a normal // comment?

[gab, 2016/08/12 23:27:46]

Sure, a /* */ comment felt more like a split from the rest to me but I guess a
// comment surrounded by empty lines works just as well.

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

[brettw, 2016/08/12 19:30:00]

This should have a reference to a bug to remove it.

[gab, 2016/08/12 23:27:46]

Done.

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

[brettw, 2016/08/12 19:30:00]

This should have a reference to a bug to remove it.

[gab, 2016/08/12 23:27:46]

Done.

+  // 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 123 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) {
+      // Confirm that the TaskScheduler's shutdown behaviors use the same

[brettw, 2016/08/12 19:30:00]

This should have a reference to a bug to remove it.

[gab, 2016/08/12 23:27:46]

Doesn't |g_all_pools_state| cover this?

+      // 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.");
 
-    create_thread_id = PrepareToStartAdditionalThreadIfHelpful();
+      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
+        // currently legitimatelly assuming 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);
+      }
+    } else {  // !REDIRECTED_TO_TASK_SCHEDULER
+      pending_tasks_.insert(sequenced);
+
+      if (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();
+  // 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(!create_thread_id);
+  } else {
+    DCHECK(sequenced_task_runner_map_.empty());
+  }
+
+  if (g_all_pools_state != AllPoolsState::REDIRECTED_TO_TASK_SCHEDULER) {
+    // 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 {
-  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)

[brettw, 2016/08/12 19:30:00]

This should have a reference to a bug to remove it.

[gab, 2016/08/12 23:27:46]

Doesn't |g_all_pools_state| cover this?

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

[brettw, 2016/08/12 19:30:00]

This should have a reference to a bug to remove it.

[gab, 2016/08/12 23:27:46]

Doesn't |g_all_pools_state| cover this?

+    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