Test SequencedWorkerPool with redirection to the 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 <unordered_map>
 #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"
@@ skipping 22 matching lines @@
 #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.
+// An enum representing the state of all pools. Any given non-test 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
+// RedirectToTaskSchedulerForProcess() 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. In a test process,
+// ResetRedirectToTaskSchedulerForProcessForTesting() causes a transition to
+// the NONE_ACTIVE state.
 // 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()
@@ skipping 292 matching lines @@
   };
 
   enum CleanupState {
     CLEANUP_REQUESTED,
     CLEANUP_STARTING,
     CLEANUP_RUNNING,
     CLEANUP_FINISHING,
     CLEANUP_DONE,
   };
 
+  struct TaskRunnerAndShutdownBehavior {
+    scoped_refptr<TaskRunner> task_runner;
+    TaskShutdownBehavior shutdown_behavior;
+  };
+
+  bool RunsTasksOnCurrentThreadAssertSynchronized() const;
+
   // 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);
+  bool PostTaskToTaskScheduler(const SequencedTask& sequenced,
+                               const TimeDelta& delay);
 
   // 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:
   //
@@ skipping 137 matching lines @@
   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_;
 
+  // The single-threaded TaskRunner used to redirect tasks to the TaskScheduler.
+  // Always null when the pool isn't single-threaded.
+  TaskRunnerAndShutdownBehavior single_threaded_task_runner_;
+
   // 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_;
+  // sequenced tasks to the TaskScheduler.
+  std::unordered_map<int, TaskRunnerAndShutdownBehavior>
+      sequenced_task_runner_map_;
+
+  // TaskScheduler TaskRunners to redirect unsequenced tasks to the
+  // TaskScheduler.
+  std::unordered_map<TaskShutdownBehavior, scoped_refptr<TaskRunner>>
+      unsequenced_task_runners_;
 
   // 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 ---------------------------------------------------------
@@ skipping 114 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;
 
+  bool task_successfully_posted = true;
+
   int create_thread_id = 0;
   {
     AutoLock lock(lock_);
 
-    if (shutdown_called_) {
+    if (shutdown_called_ &&
+        g_all_pools_state != AllPoolsState::REDIRECTED_TO_TASK_SCHEDULER) {

[gab, 2016/09/07 14:59:09]

Good catch, trunk uses atomics for this now though.

       // 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() &&
           found->second->task_shutdown_behavior() != BLOCK_SHUTDOWN) {
@@ skipping 15 matching lines @@
         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);
+      task_successfully_posted = PostTaskToTaskScheduler(sequenced, delay);
     } else {
       pending_tasks_.insert(sequenced);
 
       if (sequenced.shutdown_behavior == BLOCK_SHUTDOWN)
         blocking_shutdown_pending_task_count_++;
 
       create_thread_id = PrepareToStartAdditionalThreadIfHelpful();
     }
   }
 
@@ skipping 13 matching lines @@
     // 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;
+  return task_successfully_posted;
 }
 
-void SequencedWorkerPool::Inner::PostTaskToTaskScheduler(
-    const SequencedTask& sequenced) {
+bool SequencedWorkerPool::Inner::RunsTasksOnCurrentThreadAssertSynchronized()
+    const {
+  lock_.AssertAcquired();
+  if (g_all_pools_state == AllPoolsState::REDIRECTED_TO_TASK_SCHEDULER) {

[gab, 2016/09/07 14:59:09]

atomic

+    // TODO(fdoray): Add a special case for single-threaded pools.
+    if (!runs_tasks_on_verifier_) {
+      runs_tasks_on_verifier_ = CreateTaskRunnerWithTraits(
+          TaskTraits().WithFileIO().WithPriority(task_priority_),
+          ExecutionMode::PARALLEL);
+    }
+    return runs_tasks_on_verifier_->RunsTasksOnCurrentThread();
+  }
+  return ContainsKey(threads_, PlatformThread::CurrentId());
+}
+
+bool SequencedWorkerPool::Inner::PostTaskToTaskScheduler(
+    const SequencedTask& sequenced,
+    const TimeDelta& delay) {
   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);
+  // Find or create the TaskScheduler TaskRunner to redirect this task to.
+  scoped_refptr<TaskRunner> task_runner;
+  if (max_threads_ == 1) {

[gab, 2016/09/07 14:59:08]

I'm thinking we might just want to disallow pools with only one thread. The only
remaining ones are "WebCrypto" and image_fetcher_unittest.mm. Doing so would
much simplify the redirection and I don't think there's a valid reason to use
single-threaded SWP now that we have non-joinable Threads.

I can do this if we agree.

+    if (!single_threaded_task_runner_.task_runner) {
+      single_threaded_task_runner_.task_runner = CreateTaskRunnerWithTraits(
+          pool_traits, ExecutionMode::SINGLE_THREADED);
+      single_threaded_task_runner_.shutdown_behavior = task_shutdown_behavior;
     }
+
+    // Single-threaded pools can legitimately assume thread affinity. Disallow
+    // posting tasks with different shutdown behaviors in such pools since the
+    // TaskScheduler can't force them to run on the same thread.
+    DCHECK_EQ(task_shutdown_behavior,
+              single_threaded_task_runner_.shutdown_behavior);

[gab, 2016/09/07 14:59:08]

Different sequences on single-threaded pool could have different shutdown
behaviors though, right?

+
+    task_runner = single_threaded_task_runner_.task_runner;
+  } else if (sequenced.sequence_token_id) {
+    TaskRunnerAndShutdownBehavior& task_runner_and_shutdown_behavior =
+        sequenced_task_runner_map_[sequenced.sequence_token_id];
+    if (!task_runner_and_shutdown_behavior.task_runner) {
+      task_runner_and_shutdown_behavior.task_runner =
+          CreateTaskRunnerWithTraits(pool_traits, ExecutionMode::SEQUENCED);
+      task_runner_and_shutdown_behavior.shutdown_behavior =
+          task_shutdown_behavior;
+    }
+
+    // Posting tasks to the same sequence with different shutdown behaviors
+    // isn't supported by the TaskScheduler.
+    DCHECK_EQ(task_shutdown_behavior,
+              task_runner_and_shutdown_behavior.shutdown_behavior);
+
+    task_runner = task_runner_and_shutdown_behavior.task_runner;
+  } else {
+    scoped_refptr<TaskRunner>& task_runner_for_shutdown_behavior_ref =
+        unsequenced_task_runners_[task_shutdown_behavior];
+    if (!task_runner_for_shutdown_behavior_ref) {
+      task_runner_for_shutdown_behavior_ref =
+          CreateTaskRunnerWithTraits(pool_traits, ExecutionMode::PARALLEL);
+    }
+
+    task_runner = task_runner_for_shutdown_behavior_ref;
   }
 
-  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);
-  }
+  return task_runner->PostDelayedTask(sequenced.posted_from, sequenced.task,
+                                      delay);
 }
 
 bool SequencedWorkerPool::Inner::RunsTasksOnCurrentThread() const {
   AutoLock lock(lock_);
-  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 {
-    return ContainsKey(threads_, PlatformThread::CurrentId());
-  }
+  return RunsTasksOnCurrentThreadAssertSynchronized();
 }
 
 bool SequencedWorkerPool::Inner::IsRunningSequenceOnCurrentThread(
     SequenceToken sequence_token) const {
   AutoLock lock(lock_);
   if (g_all_pools_state == AllPoolsState::REDIRECTED_TO_TASK_SCHEDULER) {
+    if (!sequence_token.IsValid())
+      return RunsTasksOnCurrentThreadAssertSynchronized();

[danakj, 2016/09/01 23:16:34]

The non-task runner code would return false if the token was invalid wouldnt it?
Cuz it wouldnt be equal to what is currently running? Or could that one be
invalid and equal?

+
     // TODO(gab): This currently only verifies that the current thread is a

[danakj, 2016/09/01 23:16:34]

This TODO applies to the above also right

     // 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();
+           sequenced_task_runner_it->second.task_runner
+               ->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());
   }
 }
 
@@ skipping 526 matching lines @@
 scoped_refptr<SequencedWorkerPool>
 SequencedWorkerPool::GetWorkerPoolForCurrentThread() {
   Worker* worker = Worker::GetForCurrentThread();
   if (!worker)
     return nullptr;
 
   return worker->worker_pool();
 }
 
 // static
-void SequencedWorkerPool::
-    RedirectSequencedWorkerPoolsToTaskSchedulerForProcess() {
+void SequencedWorkerPool::RedirectToTaskSchedulerForProcess() {
   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;
 }
 
+// static
+void SequencedWorkerPool::ResetRedirectToTaskSchedulerForProcessForTesting() {
+  g_all_pools_state = AllPoolsState::NONE_ACTIVE;

[danakj, 2016/09/01 23:16:34]

DCHECK that it's REDIRECTED_TO_TASK_SCHEDULER currently?

[gab, 2016/09/07 14:59:08]

And use atomics now.

+}
+
 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 119 matching lines @@
 bool SequencedWorkerPool::RunsTasksOnCurrentThread() const {
   return inner_->RunsTasksOnCurrentThread();
 }
 
 bool SequencedWorkerPool::IsRunningSequenceOnCurrentThread(
     SequenceToken sequence_token) const {
   return inner_->IsRunningSequenceOnCurrentThread(sequence_token);
 }
 
 void SequencedWorkerPool::FlushForTesting() {
+  DCHECK_NE(AllPoolsState::REDIRECTED_TO_TASK_SCHEDULER, g_all_pools_state);

[gab, 2016/09/07 14:59:08]

Atomics

   inner_->CleanupForTesting();
 }
 
 void SequencedWorkerPool::SignalHasWorkForTesting() {
   inner_->SignalHasWorkForTesting();
 }
 
 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