TaskScheduler [6] SchedulerWorkerThread

base/task_scheduler/worker_thread.cc

Index: base/task_scheduler/worker_thread.cc
diff --git a/base/task_scheduler/worker_thread.cc b/base/task_scheduler/worker_thread.cc
new file mode 100644
index 0000000000000000000000000000000000000000..b76cea33a54d9bc47bb1fd35a0c733a5a6e83bdd
--- /dev/null
+++ b/base/task_scheduler/worker_thread.cc
@@ -0,0 +1,258 @@
+// 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/worker_thread.h"
+
+#include <utility>
+
+#include "base/bind.h"
+#include "base/logging.h"
+#include "base/task_scheduler/task_tracker.h"
+#include "base/task_scheduler/utils.h"
+#include "base/time/time.h"
+
+namespace base {
+namespace internal {
+
+namespace {
+
+class SchedulerSingleThreadTaskRunner : public SingleThreadTaskRunner {
+ public:
+  SchedulerSingleThreadTaskRunner(const TaskTraits& traits,
+                                  PriorityQueue* single_thread_priority_queue,
+                                  TaskTracker* task_tracker);
+
+  // SingleThreadTaskRunner:
+  bool PostNonNestableDelayedTask(const tracked_objects::Location& from_here,
+                                  const Closure& task,
+                                  TimeDelta delay) override;
+  bool PostDelayedTask(const tracked_objects::Location& from_here,
+                       const Closure& closure,
+                       TimeDelta delay) override;
+  bool RunsTasksOnCurrentThread() const override;
+
+ private:
+  ~SchedulerSingleThreadTaskRunner() override;
+
+  const TaskTraits traits_;
+  const scoped_refptr<Sequence> sequence_;
+  PriorityQueue* const priority_queue_;
+  TaskTracker* const task_tracker_;
+
+  DISALLOW_COPY_AND_ASSIGN(SchedulerSingleThreadTaskRunner);
+};
+
+SchedulerSingleThreadTaskRunner::SchedulerSingleThreadTaskRunner(
+    const TaskTraits& traits,
+    PriorityQueue* single_thread_priority_queue,
+    TaskTracker* task_tracker)
+    : traits_(traits),
+      sequence_(new Sequence),
+      priority_queue_(single_thread_priority_queue),
+      task_tracker_(task_tracker) {}
+
+bool SchedulerSingleThreadTaskRunner::PostDelayedTask(
+    const tracked_objects::Location& from_here,
+    const Closure& closure,
+    TimeDelta delay) {
+  // TODO(fdoray): Support delayed tasks.
+  DCHECK(delay.is_zero());
+  PostTaskHelper(make_scoped_ptr(new Task(from_here, closure, traits_)),
+                 sequence_, priority_queue_, task_tracker_);
+  return true;
+}
+
+bool SchedulerSingleThreadTaskRunner::RunsTasksOnCurrentThread() const {
+  // TODO(fdoray): Return true only if tasks posted may actually run on the
+  // current thread. It is valid, but not ideal, to always return true.
+  return true;

[gab, 2016/03/21 19:11:54]

I think this can be implemented relatively easily in this by:

1) Having a scoped_ptr<ThreadCheckerImpl> member in WorkerThread which is
initialized on from ThreadMain().
2) Having this SchedulerSingleThreadTaskRunner task a non-null
ThreadCheckerImpl* in constructor.
3) Verifying against it here.

This is akin to how MessageLoopTaskRunner implements it [1], but I prefer using
ThreadCheckerImpl as it essentially does the same thing while hiding the
low-level details in a single member.

Note: I'd add this to (1):
// A ThreadChecker to be used to verify that the current thread is this
// WorkerThread. Uses ThreadCheckerImpl instead of ThreadChecker to be able to
// support the non-dcheck only RunsTasksOnCurrentThread() SingleThreadTaskRunner
// API.

[1]
https://code.google.com/p/chromium/codesearch#chromium/src/base/message_loop/...

[fdoray, 2016/03/24 19:21:09]

As discussed earlier, this will be implemented with a ThreadLocalPointer. Should
I do it in this CL?

+}
+
+bool SchedulerSingleThreadTaskRunner::PostNonNestableDelayedTask(
+    const tracked_objects::Location& from_here,
+    const Closure& task,
+    TimeDelta delay) {
+  return PostDelayedTask(from_here, task, delay);

[gab, 2016/03/21 19:11:54]

Add:

// Tasks are never nested on WorkerThread.

[fdoray, 2016/03/24 19:21:10]

Done.

+}
+
+SchedulerSingleThreadTaskRunner::~SchedulerSingleThreadTaskRunner() = default;
+
+void PushSequenceInPriorityQueue(scoped_refptr<Sequence> sequence,
+                                 PriorityQueue* priority_queue) {
+  const SequenceSortKey sort_key = sequence->GetSortKey();
+  priority_queue->BeginTransaction()->Push(make_scoped_ptr(
+      new PriorityQueue::SequenceAndSortKey(std::move(sequence), sort_key)));
+}
+
+}  // namespace
+
+scoped_ptr<WorkerThread> WorkerThread::CreateWorkerThread(
+    ThreadPriority thread_priority,
+    PriorityQueue* shared_priority_queue,
+    const ReinsertSequenceCallback& reinsert_sequence_callback,
+    const BecomesIdleCallback& becomes_idle_callback,
+    TaskTracker* task_tracker) {
+  scoped_ptr<WorkerThread> worker_thread(new WorkerThread(
+      thread_priority, shared_priority_queue, reinsert_sequence_callback,
+      becomes_idle_callback, task_tracker));
+
+  if (worker_thread->thread_handle_.is_null())
+    return scoped_ptr<WorkerThread>();
+  return worker_thread;
+}
+
+WorkerThread::~WorkerThread() {
+  AutoSchedulerLock auto_lock(lock_);
+  DCHECK(should_exit_for_testing_);
+}
+
+bool WorkerThread::WakeUp() {
+  AutoSchedulerLock auto_lock(lock_);
+  if (is_awake_)
+    return false;
+  is_awake_ = true;
+  wake_up_cv_->Signal();

[gab, 2016/03/21 19:11:54]

tl;dr; no explicit request here, mostly thoughts and looking for your opinion.

I find this kind of hard to read without assumptions as nothing tells me
|wake_up_cv_| is a CV on |lock_| without double-checking the constructor.

Not sure what to do about this though... comment below about using it as the
bool solves one side of the issue on the Wait() side where it is created, but
the Signal() side still comes out of the blue.

I guess this is always a problem with CVs unless there'd be a wrapper to hold
both a lock and its corresponding CV. But if Chrome hasn't needed this yet
perhaps it's also okay for us to have them side-by-side.

EDIT: Reading the test code now which uses multiple CVs for the same lock I see
how it makes sense to decouple the two. I guess it's implicit that the CV is
based on the only Lock in this class and then its name indicates what it's used
for.

[fdoray, 2016/03/24 19:21:10]

I now use a WaitableEvent instead of a ConditionVariable. waitable_event.h: "Use
a WaitableEvent when you would otherwise use a Lock+ConditionVariable to protect
a simple boolean value."

+  return true;
+}
+
+scoped_refptr<SingleThreadTaskRunner> WorkerThread::CreateTaskRunnerWithTraits(
+    const TaskTraits& traits) {
+  return scoped_refptr<SingleThreadTaskRunner>(
+      new SchedulerSingleThreadTaskRunner(
+          traits, &single_thread_priority_queue_, task_tracker_));

[gab, 2016/03/21 19:11:54]

This is only fine because we never delete WorkerThreads outside tests (in which
no tasks are ever posted after they're deleted).

Add a comment explaining why leaking raw pointers to members is okay.

[fdoray, 2016/03/24 19:21:10]

Done.

+}
+
+void WorkerThread::JoinForTesting() {
+  {
+    AutoSchedulerLock auto_lock(lock_);
+    DCHECK(!should_exit_for_testing_);
+    should_exit_for_testing_ = true;
+  }
+  WakeUp();
+  PlatformThread::Join(thread_handle_);
+}
+
+WorkerThread::WorkerThread(
+    ThreadPriority thread_priority,
+    PriorityQueue* shared_priority_queue,
+    const ReinsertSequenceCallback& reinsert_sequence_callback,
+    const BecomesIdleCallback& becomes_idle_callback,
+    TaskTracker* task_tracker)
+    : wake_up_cv_(lock_.CreateConditionVariable()),
+      is_awake_(true),
+      should_exit_for_testing_(false),
+      single_thread_priority_queue_(
+          Bind(IgnoreResult(&WorkerThread::WakeUp), Unretained(this)),
+          shared_priority_queue),
+      shared_priority_queue_(shared_priority_queue),
+      reinsert_sequence_callback_(reinsert_sequence_callback),
+      becomes_idle_callback_(becomes_idle_callback),
+      task_tracker_(task_tracker) {
+  DCHECK(shared_priority_queue_);
+  DCHECK(!reinsert_sequence_callback_.is_null());
+  DCHECK(!becomes_idle_callback_.is_null());
+  DCHECK(task_tracker_);
+
+#if defined(OS_MACOSX)
+  // Mac only supports 2 priorities. crbug.com/554651

[gab, 2016/03/21 19:11:54]

Looks like this is fixed, I think we can remove the Mac only code here.

[fdoray, 2016/03/24 19:21:10]

Done. I pinged erikchen@ to have this landed.

+  if (thread_priority != ThreadPriority::NORMAL &&
+      thread_priority != ThreadPriority::REALTIME_AUDIO) {
+    thread_priority = ThreadPriority::NORMAL;
+  }
+#endif  // defined(OS_MACOSX)
+
+  const size_t kDefaultStackSize = 0;
+  PlatformThread::CreateWithPriority(kDefaultStackSize, this, &thread_handle_,

[gab, 2016/03/21 19:11:54]

In theory we'd need to use CreateNonJoinable() -- or introduce
CreateNonJoinableWithPriority() -- since we don't join our threads on shutdown
(right?).

We do wait for work queues to be empty though. Looks like we need to understand
the impact of a non-joinable thread.

AFAICT it only prevents use of singletons, which is probably too strict for the
way we handle things. That restrictions is probably merely so that a singleton
isn't in use while it gets destroyed (late on main thread by AtExitManager or
statically I think). In which case we could be fine enforcing ourselves that
singletons not be allowed in the scope of CONTINUE_ON_SHUTDOWN tasks, but be
okay on others.

Today, the POSIX WorkerPool uses this, but SequencedWorkerPool doesn't -- which
is probably a bug (or at least a failure to check for potential bugs w.r.t.
CONTINUE_ON_SHUTDOWN).

[fdoray, 2016/03/24 19:21:09]

We don't want to use CreateNonJoinable() because:
- We need to be able to join the thread in tests + I don't think we want a
different behavior in test vs. prod.
- It prevents the use of singletons on the thread. I think we should call
SetSingletonAllowed(false) ourselves for CONTINUE_ON_SHUTDOWN tasks (maybe in
TaskTracker::RunTask?).

+                                     thread_priority);
+}
+
+scoped_refptr<Sequence> WorkerThread::GetWork(bool* single_thread) {
+  DCHECK(single_thread);
+  *single_thread = false;
+
+  scoped_ptr<PriorityQueue::Transaction> shared_transaction(
+      shared_priority_queue_->BeginTransaction());
+  const PriorityQueue::SequenceAndSortKey shared_sequence =
+      shared_transaction->Peek();
+
+  scoped_ptr<PriorityQueue::Transaction> single_thread_transaction(
+      single_thread_priority_queue_.BeginTransaction());
+  const PriorityQueue::SequenceAndSortKey single_thread_sequence =
+      single_thread_transaction->Peek();
+
+  if (single_thread_sequence.is_null() && shared_sequence.is_null()) {
+    return scoped_refptr<Sequence>();
+  }
+
+  if (single_thread_sequence.is_null() ||
+      (!shared_sequence.is_null() &&

[gab, 2016/03/21 19:11:54]

This check is redundant. If single_thread_sequence.is_null() then it's
impossible that shared_sequence.is_null() or the condition on line 184 would
have been true.

[fdoray, 2016/03/24 19:21:10]

No. If we remove !shared_sequence.is_null(), |shared_sequence| is null and
|single_thread_sequence| is *not* null:
- Condition at line 184 is false.
- Condition at lines 188-190 is true... but should be false.

Should I make sure that the |sort_key| of a null SequenceAndSortKey is less than
any other sort key? You asked me to rely solely on is_null() instead of playing
with std::numeric_limits in another review.

[gab, 2016/04/05 23:35:20]

True, my bad, this is fine as-is nvm.

+       single_thread_sequence.sort_key < shared_sequence.sort_key)) {
+    shared_transaction->Pop();
+    return shared_sequence.sequence;
+  }
+
+  DCHECK(!single_thread_sequence.is_null());
+  single_thread_transaction->Pop();
+  *single_thread = true;
+  return single_thread_sequence.sequence;
+}
+
+void WorkerThread::WaitUntilWakeUp() {
+  AutoSchedulerLock auto_lock(lock_);
+  while (!is_awake_ && !should_exit_for_testing_)
+    wake_up_cv_->Wait();

[gab, 2016/03/21 19:11:54]

So long as WaitUntilWakeUp() is only called from one thread, i.e. single waiter
(can enforce with the ThreadCheckerImpl proposed above).

Then I think we can do the same trick we did in TaskTracker and use the CV's
existence as the bool.

I guess it depends how often we think we'll need to re-create this CV and
whether that would result in heap churn (the Shutdown case in TaskTracker was an
easy case, this one is less clear I just tend to not like things that have more
or less the same state across two members).

[fdoray, 2016/03/24 19:21:10]

I now use a WaitableEvent.

+}
+
+void WorkerThread::ThreadMain() {
+  while (!should_exit_for_testing_) {

[gab, 2016/03/21 19:11:54]

Add a comment explaining what happens in product (i.e. run out of task, sleep,
and get killed by the OS on main exit) and why this is considered preferable to
joining all threads.

[fdoray, 2016/03/24 19:21:10]

I added !task_tracker_->shutdown_completed() so that the thread exits when
shutdown completes. That way, it won't waste CPU to empty its priority queues
from remaining CONTINUE/SKIP_ON_SHUTDOWN tasks (these tasks won't run, but with
the existing code, the WorkerThread will waste time popping them from their
pqs).

+    // Get the sequence containing the next task to execute.
+    bool sequence_is_single_threaded = false;
+    scoped_refptr<Sequence> sequence = GetWork(&sequence_is_single_threaded);
+    if (sequence.get() == nullptr) {

[gab, 2016/03/21 19:11:54]

if (!sequence)

(here and below)

[fdoray, 2016/03/24 19:21:10]

Done.

+      // Mark the WorkerThread as idle.
+      {
+        AutoSchedulerLock auto_lock(lock_);
+        is_awake_ = false;
+      }
+      becomes_idle_callback_.Run(this);
+
+      // Check one more time if there is work available. If work is added to
+      // |shared_priority_queue_| after the first call to GetWork() but before

[gab, 2016/03/21 19:11:54]

Could also have been added to |single_thread_priority_queue|, no?

The way I understand this, this is:

// Check one more time if there is work available. Work could have been added
// between the first call to GetWork() and the |becomes_idle_callback_|
// invocation in which case this WorkerThread shouldn't go to sleep.

[fdoray, 2016/03/24 19:21:10]

Done. This extra check is really just for |shared_priority_queue_| because
|wake_up_event_| is always signaled when work is added to
|single_thread_priority_queue_|.

+      // the |becomes_idle_callback_| invocation, this WorkerThread should run
+      // this work.
+      sequence = GetWork(&sequence_is_single_threaded);

[gab, 2016/03/21 19:11:54]

If we do get work here don't we need to self-wake, i.e. mark |is_awake_ =
true;|?

[fdoray, 2016/03/24 19:21:10]

Done.

+
+      // Check |should_exit_for_testing_| one more time. If JoinForTesting() is
+      // called after the loop condition is checked but before |is_awake_| is
+      // updated, this WorkerThread could block on WaitUntilWakeUp() forever
+      // without this extra check.
+      if (should_exit_for_testing_)

[gab, 2016/03/21 19:11:54]

Need to lock to check this member, right?

Can we avoid having an AutoLock in product code for test-only code though?

[fdoray, 2016/03/24 19:21:10]

We no longer need this with a WaitableEvent.

+        break;
+
+      if (sequence.get() == nullptr) {
+        WaitUntilWakeUp();
+        sequence = GetWork(&sequence_is_single_threaded);

[gab, 2016/03/21 19:11:54]

Here I'd say "continue;" is easier to read, results in the same behavior but
makes one less piece in logic, i.e. "restart" instead of "re-get work and hope
to get something but we still might not".

Also "continue;" allows us to get rid of the null check on line 239 :-)

[fdoray, 2016/03/24 19:21:10]

Done.

+      }
+    }
+
+    if (sequence.get() != nullptr) {
+      // Peek the next task in the sequence and run it.
+      task_tracker_->RunTask(sequence->PeekTask());
+
+      // Pop the task from its sequence. If the sequence isn't empty, reinsert
+      // it in the appropriate PriorityQueue.
+      if (!sequence->PopTask()) {
+        if (sequence_is_single_threaded) {
+          PushSequenceInPriorityQueue(std::move(sequence),

[gab, 2016/03/21 19:11:54]

Inline this method here? It's a 2 lines helper and only used here.

[fdoray, 2016/03/24 19:21:09]

Done.

+                                      &single_thread_priority_queue_);
+        } else {
+          reinsert_sequence_callback_.Run(std::move(sequence), this);
+        }
+      }
+    }
+  }
+}
+
+}  // namespace internal
+}  // namespace base