TaskScheduler [11] Support ExecutionMode::SINGLE_THREADED.

base/task_scheduler/scheduler_worker_thread.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_thread.h"
 
 #include <stddef.h>
 
 #include <utility>
 
+#include "base/lazy_instance.h"
 #include "base/logging.h"
+#include "base/memory/ptr_util.h"
 #include "base/task_scheduler/task_tracker.h"
+#include "base/task_scheduler/utils.h"
+#include "base/threading/thread_local.h"
+#include "base/time/time.h"
 
 namespace base {
 namespace internal {
 
+namespace {
+
+// SchedulerWorkerThread that owns the current thread, if any.
+LazyInstance<ThreadLocalPointer<const SchedulerTaskExecutor>>::Leaky
+    tls_current_worker_thread = LAZY_INSTANCE_INITIALIZER;
+
+// A task runner that runs tasks on a single SchedulerWorkerThread.
+class SchedulerSingleThreadTaskRunner : public SingleThreadTaskRunner {
+ public:
+  // Constructs a SchedulerSingleThreadTaskRunner which can be used to post
+  // tasks so long as |executor| is alive.
+  // TODO(robliao): Find a concrete way to manage |executor|'s memory.
+  SchedulerSingleThreadTaskRunner(const TaskTraits& traits,
+                                  SchedulerTaskExecutor* executor,
+                                  TaskTracker* task_tracker,
+                                  DelayedTaskManager* delayed_task_manager)
+      : traits_(traits),
+        executor_(executor),
+        task_tracker_(task_tracker),
+        delayed_task_manager_(delayed_task_manager) {}
+
+  // SingleThreadTaskRunner:
+  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)),

[danakj, 2016/04/20 23:16:16]

nit: maybe this conversion from delta to ticks could be in common code (inside
PostTaskToExecutor, or whatever it calls)?

[fdoray, 2016/04/21 19:31:28]

We decided to construct the Task outside of PostTaskToExecutor to avoid passing
multiple arguments that are just forwarded to the constructor of Task.  I moved
this logic to the constructor of Task:
https://codereview.chromium.org/1906813003/

+        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_worker_thread.Get().Get() == executor_;
+  }
+
+ private:
+  ~SchedulerSingleThreadTaskRunner() 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_;
+  DelayedTaskManager* const delayed_task_manager_;
+
+  DISALLOW_COPY_AND_ASSIGN(SchedulerSingleThreadTaskRunner);
+};
+
+}  // namespace
+
 std::unique_ptr<SchedulerWorkerThread>
 SchedulerWorkerThread::CreateSchedulerWorkerThread(
     ThreadPriority thread_priority,
     Delegate* delegate,
-    TaskTracker* task_tracker) {
+    TaskTracker* task_tracker,
+    DelayedTaskManager* delayed_task_manager,
+    const PriorityQueue* predecessor_priority_queue) {
   std::unique_ptr<SchedulerWorkerThread> worker_thread(
-      new SchedulerWorkerThread(thread_priority, delegate, task_tracker));
+      new SchedulerWorkerThread(thread_priority, delegate, task_tracker,
+                                delayed_task_manager,
+                                predecessor_priority_queue));
 
   if (worker_thread->thread_handle_.is_null())
     return nullptr;
   return worker_thread;
 }
 
 SchedulerWorkerThread::~SchedulerWorkerThread() {
   DCHECK(ShouldExitForTesting());
 }
 
+scoped_refptr<SingleThreadTaskRunner>
+SchedulerWorkerThread::CreateTaskRunnerWithTraits(const TaskTraits& traits) {
+  return make_scoped_refptr(new SchedulerSingleThreadTaskRunner(
+      traits, this, task_tracker_, delayed_task_manager_));
+}
+
 void SchedulerWorkerThread::WakeUp() {
   wake_up_event_.Signal();
 }
 
 void SchedulerWorkerThread::JoinForTesting() {
   {
     AutoSchedulerLock auto_lock(should_exit_for_testing_lock_);
     should_exit_for_testing_ = true;
   }
   WakeUp();
   PlatformThread::Join(thread_handle_);
 }
 
-SchedulerWorkerThread::SchedulerWorkerThread(ThreadPriority thread_priority,
-                                             Delegate* delegate,
-                                             TaskTracker* task_tracker)
+void SchedulerWorkerThread::PostTaskWithSequence(
+    std::unique_ptr<Task> task,
+    scoped_refptr<Sequence> sequence) {
+  DCHECK(task);
+  DCHECK(sequence);
+
+  const bool sequence_was_empty = AddTaskToSequenceAndPriorityQueue(
+      std::move(task), std::move(sequence), &single_threaded_priority_queue_);
+
+  // If |sequence| wasn't empty before |task| was inserted into it, the worker
+  // thread has already been woken up to process it.
+  // TODO(fdoray): Remove the worker thread from the stack of idle threads of
+  // its parent thread pool when it is woken up to run single-threaded tasks.
+  if (sequence_was_empty)
+    WakeUp();
+}
+
+SchedulerWorkerThread::SchedulerWorkerThread(
+    ThreadPriority thread_priority,
+    Delegate* delegate,
+    TaskTracker* task_tracker,
+    DelayedTaskManager* delayed_task_manager,
+    const PriorityQueue* predecessor_priority_queue)
     : wake_up_event_(false, false),
+      single_threaded_priority_queue_(predecessor_priority_queue),
       delegate_(delegate),
-      task_tracker_(task_tracker) {
+      task_tracker_(task_tracker),
+      delayed_task_manager_(delayed_task_manager) {
   DCHECK(delegate_);
   DCHECK(task_tracker_);
+  DCHECK(delayed_task_manager_);
+  DCHECK(predecessor_priority_queue);
 
   const size_t kDefaultStackSize = 0;
   PlatformThread::CreateWithPriority(kDefaultStackSize, this, &thread_handle_,
                                      thread_priority);
 }
 
 void SchedulerWorkerThread::ThreadMain() {
   delegate_->OnMainEntry();
+  tls_current_worker_thread.Get().Set(this);
 
   // A SchedulerWorkerThread starts out sleeping.
   wake_up_event_.Wait();
 
   while (!task_tracker_->shutdown_completed() && !ShouldExitForTesting()) {
     // Get the sequence containing the next task to execute.
-    scoped_refptr<Sequence> sequence = delegate_->GetWork(this);
+    bool is_single_threaded_sequence = false;
+    scoped_refptr<Sequence> sequence = delegate_->GetWork(
+        this, &single_threaded_priority_queue_, &is_single_threaded_sequence);

[danakj, 2016/04/20 23:16:16]

Or maybe you can explain some of the design choices here.

1. I'm not clear why SchedulerWorkerThread is making a task runner instead of
SchedulerThreadPool making one that runs tasks on a given fixed worker thread.
It would just bind the SchedulerWorkerThread into the task runner in some way?

2. PostTaskWithSequence looks very similar to SchedulerThreadPool's. What would
be needed to make SchedulerThreadPool the executor for single thread tasks and
share that method? A thread* argument to PostTaskWithSequence that's null if
it's not bound to a thread?

There's no intention to use these things on a worker thread that *isn't* inside
a SchedulerThreadPool is there?

3. I guess the goal here is to try choose between this thread's queue and the
global pool queue? It feels a bit unclear the way it's being done here, passing
a pointer to "maybe use this".

What about delegate_->CheckShouldRunTask() or equivalent with the top of our
priority queue, and if it's not true, then we GetWork() for something else to
do?

Or the delegate_ could be responsible for looking at the thread's pool to choose
work and enqueue into it? As it is now, the delegate_ is only for things related
to the global thread pool, which we could preserve, it'd be nice to have the
|delegate_| name reflect that somehow so it's less hidden then? Or maybe the
methods could reflect that like GetWorkFromThreadPool and
EnqueueSequenceInThreadPool or something?

[fdoray, 2016/04/21 19:31:28]

1. SchedulerThreadPool could create the TaskRunner.

2. The methods aren't so similar. STP::PostTaskWithSequence posts a task to a
shared PQ and calls STP::WakeUpOneThread while SWT::PostTaskWithSequence posts a
task to a single-threaded PQ and calls SWT::WakeUp.

If we decide to post single-threaded tasks via STP::PostTaskWithSequence, STP
would need access to the single-threaded PQ of a SWT. We could add a public
accessor on SWT or, even better, STP could own single-threaded PQs in an
std::unordered_map<SWT*, unique_ptr<PQ>>.

Pseudo-code:

void STP::PostTaskWithSequence(
    std::unique_ptr<Task> task,
    scoped_refptr<Sequence> sequence,
    SchedulerWorkerThread* worker_thread) {
  PriorityQueue* priority_queue = worker_thread ?
      single_threaded_priority_queues_[worker_thread].get() :
      &shared_priority_queue_;
  const bool sequence_was_empty = AddTaskToSequenceAndPriorityQueue(
      std::move(task), std::move(sequence), priority_queue);

  if (sequence_was_empty) {
    if (worker_thread)
      worker_thread->WakeUp();
    else
      WakeUpOneThread();
  }
}


There is no intention to use a SchedulerWorkerThread outside of a
SchedulerThreadPool.

3. If STP owns single-threaded PQs, SWT no longer need to pass a pointer to its
single-threaded PQ to its delegate. STP::SWTDelegateImpl::GetWork can decide
what Sequence to return using the members of STP only.

CheckShouldRunTask(): That method would take a single-threaded PQ as argument
and return true if the priority of the Sequence at the top of this PQ is higher
than the priority of the Sequence at the top of the shared PQ? The problem with
that is that another worker thread could pop a Sequence between the call to
CheckShouldRunTask() and GetWork(). We prefer to have a method which atomically
checks which PQ has the Sequence with the highest priority, pops the Sequence
and returns it.

GetWorkFromThreadPool/EnqueueSequenceInThreadPool: We prefer if SWT doesn't know
that it is part of a thread pool.

 
     if (!sequence) {
       wake_up_event_.Wait();
       continue;
     }
 
     task_tracker_->RunTask(sequence->PeekTask());
 
     const bool sequence_became_empty = sequence->PopTask();
 
     // If |sequence| isn't empty immediately after the pop, enqueue it to
     // maintain the invariant that a non-empty Sequence is always referenced by
     // either a PriorityQueue or a SchedulerWorkerThread. If it is empty and
     // there are live references to it, it will be enqueued when a Task is added
     // to it. Otherwise, it will be destroyed at the end of this scope.
-    if (!sequence_became_empty)
-      delegate_->EnqueueSequence(std::move(sequence));
+    if (!sequence_became_empty) {
+      if (is_single_threaded_sequence) {
+        const auto sort_key = sequence->GetSortKey();
+        single_threaded_priority_queue_.BeginTransaction()->Push(
+            WrapUnique(new PriorityQueue::SequenceAndSortKey(
+                std::move(sequence), sort_key)));
+      } else {
+        delegate_->EnqueueSequence(std::move(sequence));
+      }
+    }
 
     // Calling WakeUp() guarantees that this SchedulerWorkerThread will run
     // Tasks from Sequences returned by the GetWork() method of |delegate_|
     // until it returns nullptr. Resetting |wake_up_event_| here doesn't break
     // this invariant and avoids a useless loop iteration before going to sleep
     // if WakeUp() is called while this SchedulerWorkerThread is awake.
     wake_up_event_.Reset();
   }
 }
 
 bool SchedulerWorkerThread::ShouldExitForTesting() const {
   AutoSchedulerLock auto_lock(should_exit_for_testing_lock_);
   return should_exit_for_testing_;
 }
 
 }  // namespace internal
 }  // namespace base