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 12 matching lines @@
 // Shared PriorityQueue of a thread's SchedulerThreadPool. Not set for threads
 // that don't belong to a SchedulerThreadPool.
 LazyInstance<ThreadLocalPointer<const PriorityQueue>>::Leaky
     tls_current_shared_priority_queue = LAZY_INSTANCE_INITIALIZER;
 
 // A task runner that runs tasks with the PARALLEL ExecutionMode.
 class SchedulerParallelTaskRunner : public TaskRunner {
  public:
   SchedulerParallelTaskRunner(const TaskTraits& traits,
                               PriorityQueue* priority_queue,
-                              TaskTracker* task_tracker)
+                              TaskTracker* task_tracker,
+                              DelayedTaskManager* delayed_task_manager)
       : traits_(traits),
         priority_queue_(priority_queue),
-        task_tracker_(task_tracker) {}
+        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 {
     // TODO(fdoray): Support delayed tasks.

[gab, 2016/04/11 18:28:30]

Remove TODO

[fdoray, 2016/04/11 19:57:06]

Delayed tasks are not supported until we have the service thread.

[gab, 2016/04/12 13:35:47]

Ah right, but from this component's POV it *is* posting to the
|delayed_task_manager_|, i.e. it won't change once we make them work so it's
weird to have the TODO here, is there a better place for it? I guess here is
good because it DCHECKs the API isn't being used in an unsupported way --
although PostDelayedTask says the delay doesn't have to be honored so we could
remove the DCHECK here and instead add the TODO on DelayedTaskManager or
something to hook it up?

[fdoray, 2016/04/12 14:43:56]

Done. Removed the DCHECK here and put it above PostReadyTasks() in
delayed_task_manager.h. With this change, delayed tasks can be posted without
failure but they will never run.

TaskRunner::PostTask: "Returns true if the task *may* be run at some point in
the future".

     DCHECK(delay.is_zero());
 
     // Post the task as part of a one-off single-task Sequence.
     return PostTaskHelper(WrapUnique(new Task(from_here, closure, traits_)),
                           make_scoped_refptr(new Sequence), priority_queue_,
-                          task_tracker_);
+                          task_tracker_, delayed_task_manager_);
   }
 
   bool RunsTasksOnCurrentThread() const override {
     return tls_current_shared_priority_queue.Get().Get() == priority_queue_;
   }
 
  private:
   ~SchedulerParallelTaskRunner() override = default;
 
   const TaskTraits traits_;
   PriorityQueue* const priority_queue_;
   TaskTracker* const task_tracker_;
+  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:
   SchedulerSequencedTaskRunner(const TaskTraits& traits,
                                PriorityQueue* priority_queue,
-                               TaskTracker* task_tracker)
+                               TaskTracker* task_tracker,
+                               DelayedTaskManager* delayed_task_manager)
       : traits_(traits),
         priority_queue_(priority_queue),
-        task_tracker_(task_tracker) {}
+        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 {
     // TODO(fdoray): Support delayed tasks.
     DCHECK(delay.is_zero());
 
     // Post the task as part of |sequence_|.
     return PostTaskHelper(WrapUnique(new Task(from_here, closure, traits_)),
-                          sequence_, priority_queue_, task_tracker_);
+                          sequence_, priority_queue_, 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_shared_priority_queue.Get().Get() == priority_queue_;
   }
 
  private:
   ~SchedulerSequencedTaskRunner() override = default;
 
   // Sequence in which all Tasks posted through this TaskRunner are inserted.
   const scoped_refptr<Sequence> sequence_ = new Sequence;
 
   const TaskTraits traits_;
   PriorityQueue* const priority_queue_;
   TaskTracker* const task_tracker_;
+  DelayedTaskManager* const delayed_task_manager_;
 
   DISALLOW_COPY_AND_ASSIGN(SchedulerSequencedTaskRunner);
 };
 
 }  // namespace
 
 class SchedulerThreadPool::SchedulerWorkerThreadDelegateImpl
     : public SchedulerWorkerThread::Delegate {
  public:
   SchedulerWorkerThreadDelegateImpl(
@@ skipping 11 matching lines @@
   SchedulerThreadPool* outer_;
   const EnqueueSequenceCallback enqueue_sequence_callback_;
 
   DISALLOW_COPY_AND_ASSIGN(SchedulerWorkerThreadDelegateImpl);
 };
 
 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;
 }
 
 SchedulerThreadPool::~SchedulerThreadPool() {
 #if DCHECK_IS_ON()
   // SchedulerThreadPool should never be deleted in production unless its
   // initialization failed.
   AutoSchedulerLock auto_lock(worker_threads_lock_);
   DCHECK(join_for_testing_returned_.IsSignaled() || worker_threads_.empty());
 #endif  // DCHECK_IS_ON()
 }
 
 scoped_refptr<TaskRunner> SchedulerThreadPool::CreateTaskRunnerWithTraits(
     const TaskTraits& traits,
     ExecutionMode execution_mode) {
   switch (execution_mode) {
     case ExecutionMode::PARALLEL:
       return make_scoped_refptr(new SchedulerParallelTaskRunner(
-          traits, &shared_priority_queue_, task_tracker_));
+          traits, &shared_priority_queue_, task_tracker_,
+          delayed_task_manager_));
 
     case ExecutionMode::SEQUENCED:
       return make_scoped_refptr(new SchedulerSequencedTaskRunner(
-          traits, &shared_priority_queue_, task_tracker_));
+          traits, &shared_priority_queue_, task_tracker_,
+          delayed_task_manager_));
 
     case ExecutionMode::SINGLE_THREADED:
       // TODO(fdoray): Support SINGLE_THREADED TaskRunners.
       NOTREACHED();
       return nullptr;
   }
 
   NOTREACHED();
   return nullptr;
 }
@@ skipping 75 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)
     : shared_priority_queue_(
           Bind(&SchedulerThreadPool::WakeUpOneThread, Unretained(this))),
       worker_threads_lock_(shared_priority_queue_.container_lock()),
       idle_worker_threads_stack_cv_for_testing_(
           worker_threads_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(worker_threads_lock_);
 
   DCHECK(worker_threads_.empty());
 
   for (size_t i = 0; i < max_threads; ++i) {
     std::unique_ptr<SchedulerWorkerThread> worker_thread =
@@ skipping 27 matching lines @@
   AutoSchedulerLock auto_lock(worker_threads_lock_);
   idle_worker_threads_stack_.push(worker_thread);
   DCHECK_LE(idle_worker_threads_stack_.size(), worker_threads_.size());
 
   if (idle_worker_threads_stack_.size() == worker_threads_.size())
     idle_worker_threads_stack_cv_for_testing_->Broadcast();
 }
 
 }  // namespace internal
 }  // namespace base