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
+      // 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)
+        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();
+    }
+
+    // If a wake up is pending and we successfully detached, somehow outer_ was

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

|outer_| x2

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

Done.

+    // able to signal us which means it probably thinks we're still alive. This
+    // is bad as it will cause the WakeUp to no-op and outer_ will be stuck
+    // forever.
+    DCHECK(!detached_worker || !IsWakeUpPending()) <<
+        "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(); }
+
+ 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;

[fdoray, 2016/06/13 20:31:17]

constexpr

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

Done.

+    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();
+  }
+
+  PlatformThreadHandle thread_handle_;
+
+  SchedulerWorkerThread* outer_;
+
+  // Event signaled to wake up this SchedulerWorkerThread.

[fdoray, 2016/06/13 20:31:17]

this Thread (or whatever the name of this class is)

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

Adjusted.

+  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,
+    InitialState initial_state) {
   std::unique_ptr<SchedulerWorkerThread> worker_thread(
       new SchedulerWorkerThread(thread_priority, std::move(delegate),
                                 task_tracker));
+  // Creation happens before any other thread can reference this one, so no
+  // synchronization is necessary.
+  if (initial_state == SchedulerWorkerThread::InitialState::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());
+  // It is unexpected for worker_ to be alive and for SchedulerWorkerThread to

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

|worker_|

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

Done.

+  // destroy since SchedulerWorkerThread owns the delegate needed by worker_.
+  // For testing, this generally means JoinForTesting was not called.
+  DCHECK(!worker_);
 }
 
 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();
+
+  worker_.reset();
+}
+
+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