Virtualize The Existence of a Scheduler Worker Thread

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/logging.h"
 #include "base/task_scheduler/task_tracker.h"
 
 namespace base {
 namespace internal {
 
+class SchedulerWorkerThread::Worker : public PlatformThread::Delegate {
+ public:
+  ~Worker() override = default;
+
+  static std::unique_ptr<Worker> Create(SchedulerWorkerThread* outer) {
+    std::unique_ptr<Worker> worker(new Worker(outer));
+    worker->Initialize();
+    if (worker->thread_handle_.is_null())
+      return nullptr;
+    return worker;
+  }
+
+  // PlatformThread::Delegate.
+  void ThreadMain() override {
+    // Set if this thread was detached.
+    std::unique_ptr<Worker> detached_worker;
+
+    outer_->delegate_->OnMainEntry(outer_);
+
+    // A SchedulerWorkerThread starts out waiting for work.
+    WaitForWork();
+
+    while (!outer_->task_tracker_->shutdown_completed() &&
+           !outer_->ShouldExitForTesting()) {
+      DCHECK(outer_);
+      // Get the sequence containing the next task to execute.
+      scoped_refptr<Sequence> sequence = outer_->delegate_->GetWork(outer_);
+      if (!sequence) {
+        if (outer_->delegate_->CanDetach(outer_)) {
+          detached_worker = outer_->Detach();
+          if (detached_worker) {
+            DCHECK_EQ(detached_worker.get(), this);
+            PlatformThread::Detach(thread_handle_);
+            outer_ = nullptr;
+            break;
+          }
+        }
+        WaitForWork();
+        continue;
+      }
+
+      outer_->task_tracker_->RunTask(sequence->PeekTask());
+
+      const bool sequence_became_empty = sequence->PopTask();
+
+      // If |sequence| isn't empty immediately after the pop, re-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
+      // andthere are live references to it, it will be enqueued when a Task is

[gab, 2016/06/10 16:15:20]

"and there"

[robliao, 2016/06/10 18:03:41]

Done.

+      // added to it. Otherwise, it will be destroyed at the end of this scope.
+      if (!sequence_became_empty)
+        outer_->delegate_->ReEnqueueSequence(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();
+    }
+
+    DCHECK(!detached_worker || !IsWakeUpPending()) <<
+        "This thread was detached and woken up at the same time.";

[gab, 2016/06/10 16:15:20]

Is this bad (i.e. someone did something wrong) or is it unexpected (i.e. this
component is written in a way that this shouldn't be possible)? Clarify in
comment.

[robliao, 2016/06/10 18:03:41]

Yep, very bad. If the thread is signaled after the detach succeeds, that means
that means SchedulerWorkerThread was able to call WakeUp on on the worker, which
should not be possible. This means a WakeUp will not be serviced.

Added a comment.

+  }
+
+  void Join() { PlatformThread::Join(thread_handle_); }
+
+  void WakeUp() { wake_up_event_.Signal(); }
+
+  bool IsWakeUpPending() { return wake_up_event_.IsSignaled(); }
+
+ private:
+  Worker(SchedulerWorkerThread* outer)
+    : outer_(outer),
+      wake_up_event_(WaitableEvent::ResetPolicy::MANUAL,
+                     WaitableEvent::InitialState::NOT_SIGNALED) {
+    DCHECK(outer_);
+  }
+
+  void Initialize() {
+    const size_t kDefaultStackSize = 0;
+    PlatformThread::CreateWithPriority(kDefaultStackSize, this,
+                                       &thread_handle_,
+                                       outer_->thread_priority_);
+  }
+
+  void WaitForWork() {
+    DCHECK(outer_);
+    const TimeDelta sleep_time = outer_->delegate_->GetSleepTimeout();
+    if (sleep_time.is_max()) {
+      // Calling TimedWait with TimeDelta::Max is not recommended per
+      // http://crbug.com/465948.
+      wake_up_event_.Wait();
+    } else {
+      wake_up_event_.TimedWait(sleep_time);
+    }
+    wake_up_event_.Reset();

[gab, 2016/06/10 16:15:20]

Why not keep this event AUTOMATIC if you're going to manually Reset() it post
wait anyways?

[robliao, 2016/06/10 18:03:41]

IsSignaled apparently resets the WaitableEvent! It was news to me too!

+  }
+
+  PlatformThreadHandle thread_handle_;
+
+  SchedulerWorkerThread* outer_;
+
+  // Event signaled to wake up this SchedulerWorkerThread.
+  WaitableEvent wake_up_event_;
+
+  DISALLOW_COPY_AND_ASSIGN(Worker);
+};
+
 std::unique_ptr<SchedulerWorkerThread> SchedulerWorkerThread::Create(
     ThreadPriority thread_priority,
     std::unique_ptr<Delegate> delegate,
-    TaskTracker* task_tracker) {
+    TaskTracker* task_tracker,
+    InitialWorkerState worker_state) {
   std::unique_ptr<SchedulerWorkerThread> worker_thread(
       new SchedulerWorkerThread(thread_priority, std::move(delegate),
                                 task_tracker));
+  // Creation is single-threaded, so no synchronization is necessary.

[gab, 2016/06/10 16:15:20]

I think what you mean is:

// Creation happens before any other thread can reference this one, so no
synchronization is necessary.

?

[robliao, 2016/06/10 18:03:41]

Yep, exactly what I mean. I was going for the one-liner comment approach.
Updated.

+  if (worker_state == SchedulerWorkerThread::InitialWorkerState::ALIVE) {
+    worker_thread->CreateWorker();
+    if (!worker_thread->worker_) {
+      return nullptr;
+    }
+  }
 
-  if (worker_thread->thread_handle_.is_null())
-    return nullptr;
   return worker_thread;
 }
 
 SchedulerWorkerThread::~SchedulerWorkerThread() {
-  DCHECK(ShouldExitForTesting());
+  DCHECK(ShouldExitForTesting() || !worker_);

[gab, 2016/06/10 16:15:20]

So I thought we said the SWT object itself would always remain alive? But I also
remember we talked about a case where we might want to delete it. I forget, can
you refresh my memory on why this is desired (and perhaps add a comment so it
sticks?)

[robliao, 2016/06/10 18:03:41]

I don't recall a case where we would want the SchedulerWorkerThread to go away.
SchedulerThreadPoolImpl has a "const" worker thread list, so it always sticks
around _except_ during tests where we tear the SchedulerThreadPoolImpl down.
In fact, if we null out worker_ on Join, we can just assert for the thread,
which is probably what we want anyways.

Added a comment and updated the DCHECK.

[gab, 2016/06/13 19:23:05]

Feels we want to keep the ShouldExitForTesting() DCHECK as well though?

Shouldn't it be DCHECK(ShouldExitForTesting() && !worker_) in fact?

[robliao, 2016/06/13 23:23:28]

It seemed redundant to me as the goal of ShouldExitForTesting was to help make
sure the associated thread is gone before we destroy SchedulerWorkerThread
(leaking resources).

If you folks feel strongly about this, I can put it back in.

[gab, 2016/06/15 19:40:42]

Wasn't it also because the rest of the scheduler assumes the workers to always
be around? Or is that an impl detail of the higher-level components of the
scheduler?

 }
 
 void SchedulerWorkerThread::WakeUp() {
-  wake_up_event_.Signal();
+  AutoSchedulerLock auto_lock(worker_lock_);
+  if (!worker_)
+    CreateWorkerAssertSynchronized();
+
+  if (worker_)
+    worker_->WakeUp();
 }
 
 void SchedulerWorkerThread::JoinForTesting() {
   {
     AutoSchedulerLock auto_lock(should_exit_for_testing_lock_);
     should_exit_for_testing_ = true;
   }
   WakeUp();
-  PlatformThread::Join(thread_handle_);
+
+  // Normally holding a lock and joining is dangerous. However, since this is
+  // only for testing, we're okay since the only scenario that could impact this
+  // is a call to Detach, which is disallowed by having the delegate always
+  // return false for the CanDetach call.
+  AutoSchedulerLock auto_lock(worker_lock_);
+  if (worker_)
+    worker_->Join();
+}
+
+bool SchedulerWorkerThread::WorkerAliveForTesting() const {
+  AutoSchedulerLock auto_lock(worker_lock_);
+  return !!worker_;
 }
 
 SchedulerWorkerThread::SchedulerWorkerThread(ThreadPriority thread_priority,
                                              std::unique_ptr<Delegate> delegate,
                                              TaskTracker* task_tracker)
-    : wake_up_event_(WaitableEvent::ResetPolicy::AUTOMATIC,
-                     WaitableEvent::InitialState::NOT_SIGNALED),
+    : thread_priority_(thread_priority),
       delegate_(std::move(delegate)),
       task_tracker_(task_tracker) {
   DCHECK(delegate_);
   DCHECK(task_tracker_);
-
-  const size_t kDefaultStackSize = 0;
-  PlatformThread::CreateWithPriority(kDefaultStackSize, this, &thread_handle_,
-                                     thread_priority);
 }
 
-void SchedulerWorkerThread::ThreadMain() {
-  delegate_->OnMainEntry(this);
+std::unique_ptr<SchedulerWorkerThread::Worker> SchedulerWorkerThread::Detach() {
+  DCHECK(!ShouldExitForTesting()) << "Worker was already joined";
+  AutoSchedulerLock auto_lock(worker_lock_);
+  // If a wakeup is pending, then a WakeUp came in while we were deciding to
+  // detach. This means we can't go away anymore since we would break the
+  // guarantee that we call GetWork after a successful wakeup.
+  return worker_->IsWakeUpPending() ? nullptr : std::move(worker_);
+}
 
-  // A SchedulerWorkerThread starts out sleeping.
-  wake_up_event_.Wait();
+void SchedulerWorkerThread::CreateWorker() {
+  worker_ = Worker::Create(this);
+}
 
-  while (!task_tracker_->shutdown_completed() && !ShouldExitForTesting()) {
-    // Get the sequence containing the next task to execute.
-    scoped_refptr<Sequence> sequence = delegate_->GetWork(this);
-
-    if (!sequence) {
-      TimeDelta sleep_time = delegate_->GetSleepTimeout();
-      if (sleep_time.is_max()) {
-        // Calling TimedWait with TimeDelta::Max is not recommended per
-        // http://crbug.com/465948.
-        wake_up_event_.Wait();
-      } else {
-        wake_up_event_.TimedWait(sleep_time);
-      }
-      continue;
-    }
-
-    task_tracker_->RunTask(sequence->PeekTask());
-
-    const bool sequence_became_empty = sequence->PopTask();
-
-    // If |sequence| isn't empty immediately after the pop, re-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_->ReEnqueueSequence(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();
-  }
+void SchedulerWorkerThread::CreateWorkerAssertSynchronized() {
+  worker_lock_.AssertAcquired();
+  CreateWorker();
 }
 
 bool SchedulerWorkerThread::ShouldExitForTesting() const {
   AutoSchedulerLock auto_lock(should_exit_for_testing_lock_);
   return should_exit_for_testing_;
 }
 
 }  // namespace internal
 }  // namespace base