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 138 matching lines @@
 
   // 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_;
+  // 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) {
       // 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) {
+    // 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();
+  } else {

[robliao, 2016/08/29 23:28:50]

Nit: Remove else

[fdoray, 2016/08/30 17:20:49]

Done.

+    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;
+  // Find or create the TaskScheduler TaskRunner to redirect this task to.
+  scoped_refptr<TaskRunner> task_runner;
   if (sequenced.sequence_token_id) {
-    sequenced_task_runner =
-        &sequenced_task_runner_map_[sequenced.sequence_token_id];
-    if (!*sequenced_task_runner) {
+    TaskRunnerAndShutdownBehavior& task_runner_and_shutdown_behavior =
+        sequenced_task_runner_map_[sequenced.sequence_token_id];
+    if (!task_runner_and_shutdown_behavior.task_runner) {
       const ExecutionMode execution_mode =
           max_threads_ == 1U ? ExecutionMode::SINGLE_THREADED
                              : ExecutionMode::SEQUENCED;
-      *sequenced_task_runner =
+      task_runner_and_shutdown_behavior.task_runner =
+          CreateTaskRunnerWithTraits(pool_traits, execution_mode);
+      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) {
+      const ExecutionMode execution_mode = max_threads_ == 1U
+                                               ? ExecutionMode::SINGLE_THREADED
+                                               : ExecutionMode::PARALLEL;
+      task_runner_for_shutdown_behavior_ref =
           CreateTaskRunnerWithTraits(pool_traits, execution_mode);
     }
+
+    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);
-  }
+  // Single-threaded pools can legitimatelly assume thread affinity. Disallow

[robliao, 2016/08/29 23:28:50]

Nit: s/legitimatelly/legitimately/

[fdoray, 2016/08/30 17:20:49]

Done.

+  // posting tasks with different sequence tokens or shutdown behaviors in such
+  // pools since the TaskScheduler can't force them to run on the same thread.
+  DCHECK(max_threads_ > 1 ||
+         (unsequenced_task_runners_.size() +

[robliao, 2016/08/29 23:28:50]

A caller could plausibly call GetTaskRunnerWithShutdownBehavior with the same
shutdown behavior. Do we want to disallow those?

[fdoray, 2016/08/30 17:20:49]

No, we shouldn't disallow that. Done.

+          sequenced_task_runner_map_.size()) == 1);
+
+  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();
+
     // 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();
+           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;
+}
+
 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);
   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