TaskScheduler [9] Delayed Task Manager

base/task_scheduler/scheduler_thread_pool.cc

 // Copyright 2016 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/task_scheduler/scheduler_thread_pool.h"
 
 #include <utility>
 
 #include "base/bind.h"
 #include "base/bind_helpers.h"
@@ skipping 13 matching lines @@
 LazyInstance<ThreadLocalPointer<const SchedulerTaskExecutor>>::Leaky
     tls_current_thread_pool = LAZY_INSTANCE_INITIALIZER;
 
 // A task runner that runs tasks with the PARALLEL ExecutionMode.
 class SchedulerParallelTaskRunner : public TaskRunner {
  public:
   // Constructs a SchedulerParallelTaskRunner which can be used to post tasks so
   // long as |executor| is alive.
   // TODO(robliao): Find a concrete way to manage |executor|'s memory.
   SchedulerParallelTaskRunner(const TaskTraits& traits,
+                              SchedulerTaskExecutor* executor,
                               TaskTracker* task_tracker,
-                              SchedulerTaskExecutor* executor)
-      : traits_(traits), task_tracker_(task_tracker), executor_(executor) {}
+                              DelayedTaskManager* delayed_task_manager)
+      : traits_(traits),
+        executor_(executor),
+        task_tracker_(task_tracker),
+        delayed_task_manager_(delayed_task_manager) {}
 
   // TaskRunner:
   bool PostDelayedTask(const tracked_objects::Location& from_here,
                        const Closure& closure,
                        TimeDelta delay) override {
     // Post the task as part of a one-off single-task Sequence.
     return PostTaskToExecutor(
         WrapUnique(
             new Task(from_here, closure, traits_,
                      delay.is_zero() ? TimeTicks() : TimeTicks::Now() + delay)),
-        make_scoped_refptr(new Sequence), executor_, task_tracker_);
+        make_scoped_refptr(new Sequence), executor_, task_tracker_,
+        delayed_task_manager_);
   }
 
   bool RunsTasksOnCurrentThread() const override {
     return tls_current_thread_pool.Get().Get() == executor_;
   }
 
  private:
   ~SchedulerParallelTaskRunner() override = default;
 
   const TaskTraits traits_;
+  SchedulerTaskExecutor* const executor_;
   TaskTracker* const task_tracker_;
-  SchedulerTaskExecutor* const executor_;
+  DelayedTaskManager* const delayed_task_manager_;
 
   DISALLOW_COPY_AND_ASSIGN(SchedulerParallelTaskRunner);
 };
 
 // A task runner that runs tasks with the SEQUENCED ExecutionMode.
 class SchedulerSequencedTaskRunner : public SequencedTaskRunner {
  public:
   // Constructs a SchedulerParallelTaskRunner which can be used to post tasks so
   // long as |executor| is alive.
   // TODO(robliao): Find a concrete way to manage |executor|'s memory.
   SchedulerSequencedTaskRunner(const TaskTraits& traits,
+                               SchedulerTaskExecutor* executor,
                                TaskTracker* task_tracker,
-                               SchedulerTaskExecutor* executor)
-      : traits_(traits), task_tracker_(task_tracker), executor_(executor) {}
+                               DelayedTaskManager* delayed_task_manager)
+      : traits_(traits),
+        executor_(executor),
+        task_tracker_(task_tracker),
+        delayed_task_manager_(delayed_task_manager) {}
 
   // SequencedTaskRunner:
   bool PostDelayedTask(const tracked_objects::Location& from_here,
                        const Closure& closure,
                        TimeDelta delay) override {
     // Post the task as part of |sequence|.
     return PostTaskToExecutor(
         WrapUnique(
             new Task(from_here, closure, traits_,
                      delay.is_zero() ? TimeTicks() : TimeTicks::Now() + delay)),
-        sequence_, executor_, task_tracker_);
+        sequence_, executor_, task_tracker_, delayed_task_manager_);
   }
 
   bool PostNonNestableDelayedTask(const tracked_objects::Location& from_here,
                                   const Closure& closure,
                                   base::TimeDelta delay) override {
     // Tasks are never nested within the task scheduler.
     return PostDelayedTask(from_here, closure, delay);
   }
 
   bool RunsTasksOnCurrentThread() const override {
     return tls_current_thread_pool.Get().Get() == executor_;
   }
 
  private:
   ~SchedulerSequencedTaskRunner() override = default;
 
   // Sequence for all Tasks posted through this TaskRunner.
   const scoped_refptr<Sequence> sequence_ = new Sequence;
 
   const TaskTraits traits_;
+  SchedulerTaskExecutor* const executor_;
   TaskTracker* const task_tracker_;
-  SchedulerTaskExecutor* const executor_;
+  DelayedTaskManager* const delayed_task_manager_;
 
   DISALLOW_COPY_AND_ASSIGN(SchedulerSequencedTaskRunner);
 };
 
 }  // namespace
 
 class SchedulerThreadPool::SchedulerWorkerThreadDelegateImpl
     : public SchedulerWorkerThread::Delegate {
  public:
   SchedulerWorkerThreadDelegateImpl(
@@ skipping 17 matching lines @@
 SchedulerThreadPool::~SchedulerThreadPool() {
   // SchedulerThreadPool should never be deleted in production unless its
   // initialization failed.
   DCHECK(join_for_testing_returned_.IsSignaled() || worker_threads_.empty());
 }
 
 std::unique_ptr<SchedulerThreadPool> SchedulerThreadPool::CreateThreadPool(
     ThreadPriority thread_priority,
     size_t max_threads,
     const EnqueueSequenceCallback& enqueue_sequence_callback,
-    TaskTracker* task_tracker) {
-  std::unique_ptr<SchedulerThreadPool> thread_pool(
-      new SchedulerThreadPool(enqueue_sequence_callback, task_tracker));
+    TaskTracker* task_tracker,
+    DelayedTaskManager* delayed_task_manager) {
+  std::unique_ptr<SchedulerThreadPool> thread_pool(new SchedulerThreadPool(
+      enqueue_sequence_callback, task_tracker, delayed_task_manager));
   if (thread_pool->Initialize(thread_priority, max_threads))
     return thread_pool;
   return nullptr;
 }
 
 scoped_refptr<TaskRunner> SchedulerThreadPool::CreateTaskRunnerWithTraits(
     const TaskTraits& traits,
     ExecutionMode execution_mode) {
   switch (execution_mode) {
     case ExecutionMode::PARALLEL:
-      return make_scoped_refptr(
-          new SchedulerParallelTaskRunner(traits, task_tracker_, this));
+      return make_scoped_refptr(new SchedulerParallelTaskRunner(
+          traits, this, task_tracker_, delayed_task_manager_));
 
     case ExecutionMode::SEQUENCED:
-      return make_scoped_refptr(
-          new SchedulerSequencedTaskRunner(traits, task_tracker_, this));
+      return make_scoped_refptr(new SchedulerSequencedTaskRunner(
+          traits, this, task_tracker_, delayed_task_manager_));
 
     case ExecutionMode::SINGLE_THREADED:
       // TODO(fdoray): Support SINGLE_THREADED TaskRunners.
       NOTREACHED();
       return nullptr;
   }
 
   NOTREACHED();
   return nullptr;
 }
@@ skipping 88 matching lines @@
   return sequence_and_sort_key.sequence;
 }
 
 void SchedulerThreadPool::SchedulerWorkerThreadDelegateImpl::EnqueueSequence(
     scoped_refptr<Sequence> sequence) {
   enqueue_sequence_callback_.Run(std::move(sequence));
 }
 
 SchedulerThreadPool::SchedulerThreadPool(
     const EnqueueSequenceCallback& enqueue_sequence_callback,
-    TaskTracker* task_tracker)
+    TaskTracker* task_tracker,
+    DelayedTaskManager* delayed_task_manager)
     : idle_worker_threads_stack_lock_(shared_priority_queue_.container_lock()),
       idle_worker_threads_stack_cv_for_testing_(
           idle_worker_threads_stack_lock_.CreateConditionVariable()),
       join_for_testing_returned_(true, false),
       worker_thread_delegate_(
           new SchedulerWorkerThreadDelegateImpl(this,
                                                 enqueue_sequence_callback)),
-      task_tracker_(task_tracker) {
+      task_tracker_(task_tracker),
+      delayed_task_manager_(delayed_task_manager) {
   DCHECK(task_tracker_);
+  DCHECK(delayed_task_manager_);
 }
 
 bool SchedulerThreadPool::Initialize(ThreadPriority thread_priority,
                                      size_t max_threads) {
   AutoSchedulerLock auto_lock(idle_worker_threads_stack_lock_);
 
   DCHECK(worker_threads_.empty());
 
   for (size_t i = 0; i < max_threads; ++i) {
     std::unique_ptr<SchedulerWorkerThread> worker_thread =
@@ skipping 30 matching lines @@
   if (idle_worker_threads_stack_.empty())
     return nullptr;
 
   auto worker_thread = idle_worker_threads_stack_.top();
   idle_worker_threads_stack_.pop();
   return worker_thread;
 }
 
 }  // namespace internal
 }  // namespace base