TaskScheduler: Replace the SchedulerServiceThread with a base::Thread.

base/task_scheduler/scheduler_worker_pool_impl.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_worker_pool_impl.h"
 
 #include <stddef.h>
 
 #include <algorithm>
 #include <utility>
 
 #include "base/atomicops.h"
 #include "base/bind.h"
 #include "base/bind_helpers.h"
 #include "base/lazy_instance.h"
 #include "base/memory/ptr_util.h"
 #include "base/metrics/histogram.h"
 #include "base/sequence_token.h"
 #include "base/sequenced_task_runner.h"
 #include "base/single_thread_task_runner.h"
 #include "base/strings/stringprintf.h"
-#include "base/task_scheduler/delayed_task_manager.h"
 #include "base/task_scheduler/task_tracker.h"
 #include "base/threading/platform_thread.h"
 #include "base/threading/thread_local.h"
 #include "base/threading/thread_restrictions.h"
 #include "base/time/time.h"
 
 namespace base {
 namespace internal {
 
 namespace {
@@ skipping 25 matching lines @@
     DCHECK(worker_pool_);
   }
 
   // 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 worker_pool_->PostTaskWithSequence(
         MakeUnique<Task>(from_here, closure, traits_, delay),
-        make_scoped_refptr(new Sequence), nullptr);
+        make_scoped_refptr(new Sequence), nullptr, delay);
   }
 
   bool RunsTasksOnCurrentThread() const override {
     return tls_current_worker_pool.Get().Get() == worker_pool_;
   }
 
  private:
   ~SchedulerParallelTaskRunner() override = default;
 
   const TaskTraits traits_;
@@ skipping 16 matching lines @@
 
   // SequencedTaskRunner:
   bool PostDelayedTask(const tracked_objects::Location& from_here,
                        const Closure& closure,
                        TimeDelta delay) override {
     std::unique_ptr<Task> task(new Task(from_here, closure, traits_, delay));
     task->sequenced_task_runner_ref = this;
 
     // Post the task as part of |sequence_|.
     return worker_pool_->PostTaskWithSequence(std::move(task), sequence_,
-                                              nullptr);
+                                              nullptr, delay);
   }
 
   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 {
@@ skipping 64 matching lines @@
 
   // SingleThreadTaskRunner:
   bool PostDelayedTask(const tracked_objects::Location& from_here,
                        const Closure& closure,
                        TimeDelta delay) override {
     std::unique_ptr<Task> task(new Task(from_here, closure, traits_, delay));
     task->single_thread_task_runner_ref = this;
 
     // Post the task to be executed by |worker_| as part of |sequence_|.
     return worker_pool_->PostTaskWithSequence(std::move(task), sequence_,
-                                              worker_);
+                                              worker_, delay);
   }
 
   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 {
@@ skipping 90 matching lines @@
   // SchedulerWorkerPool should never be deleted in production unless its
   // initialization failed.
   DCHECK(join_for_testing_returned_.IsSignaled() || workers_.empty());
 }
 
 // static
 std::unique_ptr<SchedulerWorkerPoolImpl> SchedulerWorkerPoolImpl::Create(
     const SchedulerWorkerPoolParams& params,
     const ReEnqueueSequenceCallback& re_enqueue_sequence_callback,
     TaskTracker* task_tracker,
-    DelayedTaskManager* delayed_task_manager) {
+    scoped_refptr<TaskRunner> service_thread_task_runner) {
   std::unique_ptr<SchedulerWorkerPoolImpl> worker_pool(
-      new SchedulerWorkerPoolImpl(params.name(),
-                                  params.io_restriction(),
-                                  params.suggested_reclaim_time(),
-                                  task_tracker, delayed_task_manager));
+      new SchedulerWorkerPoolImpl(params.name(), params.io_restriction(),
+                                  params.suggested_reclaim_time(), task_tracker,
+                                  std::move(service_thread_task_runner)));
   if (worker_pool->Initialize(params.priority_hint(), params.max_threads(),
                               re_enqueue_sequence_callback)) {
     return worker_pool;
   }
   return nullptr;
 }
 
 void SchedulerWorkerPoolImpl::WaitForAllWorkersIdleForTesting() {
   AutoSchedulerLock auto_lock(idle_workers_stack_lock_);
   while (idle_workers_stack_.Size() < workers_.size())
@@ skipping 58 matching lines @@
   // we did wake up another worker, we would waste resources by having more
   // workers trying to get a Sequence from |shared_priority_queue_| than the
   // number of Sequences in it.
   if (tls_current_worker_pool.Get().Get() != this)
     WakeUpOneWorker();
 }
 
 bool SchedulerWorkerPoolImpl::PostTaskWithSequence(
     std::unique_ptr<Task> task,
     scoped_refptr<Sequence> sequence,
-    SchedulerWorker* worker) {
+    SchedulerWorker* worker,
+    TimeDelta delay) {

[robliao, 2016/10/12 20:55:45]

I would be okay with storing the delay on the task now that we actually need it.
This would simplify the signature of PostTaskWithSequence. This has the benefit
of keeping all topical information like delay and delayed_run_time_ on the task,
removing the need for the DCHECK_EQ below.

Aside: Unsurprisingly, MessageLoop converts this delay into a TimeTicks later
on. Might be worthwhile to consider how we can pipe TimeTicks all the way
through.

[fdoray, 2016/10/13 13:51:29]

Done. Stored the delay with Task and added TODO.

   DCHECK(task);
   DCHECK(sequence);
   DCHECK(!worker || ContainsWorker(workers_, worker));
+  DCHECK_EQ(delay.is_zero(), task->delayed_run_time.is_null());
 
   if (!task_tracker_->WillPostTask(task.get()))
     return false;
 
-  if (task->delayed_run_time.is_null()) {
+  if (delay.is_zero()) {
     PostTaskWithSequenceNow(std::move(task), std::move(sequence), worker);
   } else {
-    delayed_task_manager_->AddDelayedTask(std::move(task), std::move(sequence),
-                                          worker, this);
+    service_thread_task_runner_->PostDelayedTask(
+        FROM_HERE, Bind(&SchedulerWorkerPoolImpl::PostTaskWithSequenceNow,
+                        Unretained(this), Passed(std::move(task)),
+                        std::move(sequence), worker),
+        delay);
   }
 
   return true;
 }
 
 void SchedulerWorkerPoolImpl::PostTaskWithSequenceNow(
     std::unique_ptr<Task> task,
     scoped_refptr<Sequence> sequence,
     SchedulerWorker* worker) {
   DCHECK(task);
   DCHECK(sequence);
   DCHECK(!worker || ContainsWorker(workers_, worker));
 
   // Confirm that |task| is ready to run (its delayed run time is either null or
   // in the past).
-  DCHECK_LE(task->delayed_run_time, delayed_task_manager_->Now());
+  DCHECK_LE(task->delayed_run_time, TimeTicks::Now());
 
   // Because |worker| belongs to this worker pool, we know that the type
   // of its delegate is SchedulerWorkerDelegateImpl.
   PriorityQueue* const priority_queue =
       worker
           ? static_cast<SchedulerWorkerDelegateImpl*>(worker->delegate())
                 ->single_threaded_priority_queue()
           : &shared_priority_queue_;
   DCHECK(priority_queue);
 
@@ skipping 222 matching lines @@
       !subtle::NoBarrier_Load(&num_single_threaded_runners_) &&
       outer_->CanWorkerDetachForTesting();
   return can_detach;
 }
 
 SchedulerWorkerPoolImpl::SchedulerWorkerPoolImpl(
     StringPiece name,
     SchedulerWorkerPoolParams::IORestriction io_restriction,
     const TimeDelta& suggested_reclaim_time,
     TaskTracker* task_tracker,
-    DelayedTaskManager* delayed_task_manager)
+    scoped_refptr<TaskRunner> service_thread_task_runner)
     : name_(name.as_string()),
       io_restriction_(io_restriction),
       suggested_reclaim_time_(suggested_reclaim_time),
       idle_workers_stack_lock_(shared_priority_queue_.container_lock()),
       idle_workers_stack_cv_for_testing_(
           idle_workers_stack_lock_.CreateConditionVariable()),
       join_for_testing_returned_(WaitableEvent::ResetPolicy::MANUAL,
                                  WaitableEvent::InitialState::NOT_SIGNALED),
 #if DCHECK_IS_ON()
       workers_created_(WaitableEvent::ResetPolicy::MANUAL,
@@ skipping 10 matching lines @@
       // expected to run between zero and a few tens of tasks between waits.
       // When it runs more than 100 tasks, there is no need to know the exact
       // number of tasks that ran.
       num_tasks_between_waits_histogram_(Histogram::FactoryGet(
           kNumTasksBetweenWaitsHistogramPrefix + name_ + kPoolNameSuffix,
           1,
           100,
           50,
           HistogramBase::kUmaTargetedHistogramFlag)),
       task_tracker_(task_tracker),
-      delayed_task_manager_(delayed_task_manager) {
+      service_thread_task_runner_(std::move(service_thread_task_runner)) {
   DCHECK(task_tracker_);
-  DCHECK(delayed_task_manager_);
+  DCHECK(service_thread_task_runner_);
 }
 
 bool SchedulerWorkerPoolImpl::Initialize(
     ThreadPriority priority_hint,
     size_t max_threads,
     const ReEnqueueSequenceCallback& re_enqueue_sequence_callback) {
   AutoSchedulerLock auto_lock(idle_workers_stack_lock_);
 
   DCHECK(workers_.empty());
 
@@ skipping 53 matching lines @@
   AutoSchedulerLock auto_lock(idle_workers_stack_lock_);
   idle_workers_stack_.Remove(worker);
 }
 
 bool SchedulerWorkerPoolImpl::CanWorkerDetachForTesting() {
   return !worker_detachment_disallowed_.IsSet();
 }
 
 }  // namespace internal
 }  // namespace base