Virtualize The Existence of a Scheduler Worker Thread

base/task_scheduler/scheduler_worker_thread.cc

Index: base/task_scheduler/scheduler_worker_thread.cc
diff --git a/base/task_scheduler/scheduler_worker_thread.cc b/base/task_scheduler/scheduler_worker_thread.cc
index afc70ec360ba136aee7b0e1dc719facaf678d284..31b4f2f03cbe08b4e6ab25fd8b90a8220feace74 100644
--- a/base/task_scheduler/scheduler_worker_thread.cc
+++ b/base/task_scheduler/scheduler_worker_thread.cc
@@ -14,25 +14,137 @@
 namespace base {
 namespace internal {
 
+class SchedulerWorkerThread::Worker : public PlatformThread::Delegate {

[fdoray, 2016/06/08 17:51:48]

s/::Worker/::Thread/ ?

[robliao, 2016/06/08 19:00:08]

I wanted to avoid SchedulerWorkerThread::Thread. If you folks are okay with
Thread, I'm happy to rename this.

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

Or we could have SchedulerWorker and SchedulerWorker::Thread since the former is
no longer a thread per se?

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

Happy to perform the rename. I'll do that as the very last thing to keep the
comment train preserved.

+ public:
+  ~Worker() override = default;
+
+  static std::unique_ptr<Worker> Create(SchedulerWorkerThread* outer,
+                                        ThreadPriority thread_priority) {
+    std::unique_ptr<Worker> worker(new Worker(outer));
+    worker->Initialize(thread_priority);

[fdoray, 2016/06/08 17:51:48]

Could read priority from |outer|.

[robliao, 2016/06/08 19:00:08]

Indeed! Done.

+    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 sleeping.
+    wake_up_event_.Wait();

[fdoray, 2016/06/08 17:51:48]

GetSleepTimeout()? Otherwise, a thread that is never woken up will never detach.

[robliao, 2016/06/08 19:00:08]

Done. 

Huh, interesting.  This actually means that GetSleepTimeout() is part of the
initial run instead of not idle. Seems okay now, which impacts unit tests. We
should probably reconsider putting DCHECKs in delegates if we don't actually
care about the call ordering. (SchedulerWorkerThreadStackUnittest had to be
updated for this).

+
+    while (outer_ && !outer_->task_tracker_->shutdown_completed() &&

[fdoray, 2016/06/08 17:51:48]

How can |outer_| be nullptr here? You "break;" right after setting is to
nullptr.
DCHECK(outer_) inside the loop?

[robliao, 2016/06/08 19:00:08]

Indeed. I left this over from an earlier revision w/o the break.

+           !outer_->ShouldExitForTesting()) {
+      // 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;

[fdoray, 2016/06/08 17:51:48]

Is there a reason why this has to be set to nullptr?

I would change
SchedulerWorkerThread* outer_; to
SchedulerWorkerThread* const outer_;

[robliao, 2016/06/08 19:00:08]

This ensures that two workers aren't talking to a SchedulerWorkerThread at the
same time.

+            break;
+          }
+        }
+        TimeDelta sleep_time = outer_->delegate_->GetSleepTimeout();

[fdoray, 2016/06/08 17:51:48]

const TimeDelta sleep_time = 

[robliao, 2016/06/08 19:00:08]

Done.

+        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;
+      }
+
+      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
+      // 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 || !wake_up_event_.IsSignaled()) <<

[fdoray, 2016/06/08 17:51:48]

!IsWakeUpPending()

[robliao, 2016/06/08 19:00:08]

Done.

+        "This thread was detached and woken up at the same time.";
+  }
+
+  void Join() { PlatformThread::Join(thread_handle_); }
+
+  void WakeUp() { wake_up_event_.Signal(); }
+
+  bool IsWakeUpPending() { return wake_up_event_.IsSignaled(); }

[fdoray, 2016/06/08 17:51:48]

IsSignaled() resets the event...
https://cs.chromium.org/chromium/src/base/synchronization/waitable_event.h?q=...

[robliao, 2016/06/08 19:00:08]

Nice catch. That certainly seems unexpected. Changed to manual and we reset on
wake.

+
+ private:
+  Worker(SchedulerWorkerThread* outer)
+    : outer_(outer),
+      wake_up_event_(WaitableEvent::ResetPolicy::AUTOMATIC,
+                     WaitableEvent::InitialState::NOT_SIGNALED) {
+    DCHECK(outer_);
+  }
+
+  void Initialize(ThreadPriority thread_priority) {
+    const size_t kDefaultStackSize = 0;
+    PlatformThread::CreateWithPriority(kDefaultStackSize, this,
+                                       &thread_handle_, thread_priority);
+  }
+
+  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.
+  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_);
 }
 
 void SchedulerWorkerThread::WakeUp() {
-  wake_up_event_.Signal();
+  AutoSchedulerLock auto_lock(worker_lock_);
+  if (!worker_) {
+    CreateWorkerAssertSynchronized();
+  }
+  worker_->WakeUp();
 }
 
 void SchedulerWorkerThread::JoinForTesting() {
@@ -41,65 +153,45 @@ void SchedulerWorkerThread::JoinForTesting() {
     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() {
+  return !!worker_;

[fdoray, 2016/06/08 17:51:48]

AutoSchedulerLock auto_lock(worker_lock_);

[robliao, 2016/06/08 19:00:08]

Done.

 }
 
 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);
-
-  // 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);
-
-    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();
+std::unique_ptr<SchedulerWorkerThread::Worker> SchedulerWorkerThread::Detach() {

[fdoray, 2016/06/08 17:51:48]

DCHECK(!ShouldExitForTesting()); before acquiring |worker_lock_|

[robliao, 2016/06/08 19:00:08]

Done.

+  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 a single threaded task
+  // could have woken us up.

[fdoray, 2016/06/08 17:51:48]

It should be the responsibility of the delegate to prevent the thread from
detaching if single-threaded task runners are associated with the
SchedulerWorkerThread. The real reason why we don't want to detach is to avoid
breaking the guarantee that GetWork() is always called after a call to WakeUp().

[robliao, 2016/06/08 19:00:08]

Added a note about single-threaded work to CanDetach and updated this comment.
This can also happen if the thread is idle and suddenly gets scheduled to run
work.

+  return worker_->IsWakeUpPending() ? nullptr : std::move(worker_);
+}
 
-    // 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));
+void SchedulerWorkerThread::CreateWorker() {
+  worker_ = Worker::Create(this, thread_priority_);
+}
 
-    // 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 {