SchedulerWorker Refcounting for Destruction in Production

base/task_scheduler/scheduler_worker.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.h"
 
 #include <stddef.h>
 
 #include <utility>
 
 #include "base/logging.h"
 #include "base/memory/ptr_util.h"
 #include "base/task_scheduler/task_tracker.h"
 
 #if defined(OS_MACOSX)
 #include "base/mac/scoped_nsautorelease_pool.h"
 #elif defined(OS_WIN)
 #include "base/win/scoped_com_initializer.h"
 #endif
 
 namespace base {
 namespace internal {
 
 class SchedulerWorker::Thread : public PlatformThread::Delegate {
  public:
   ~Thread() override = default;
 
-  static std::unique_ptr<Thread> Create(SchedulerWorker* outer) {
-    std::unique_ptr<Thread> thread(new Thread(outer));
+  static std::unique_ptr<Thread> Create(scoped_refptr<SchedulerWorker> outer) {
+    std::unique_ptr<Thread> thread(new Thread(std::move(outer)));
     thread->Initialize();
     if (thread->thread_handle_.is_null())
       return nullptr;
     return thread;
   }
 
   // PlatformThread::Delegate.
   void ThreadMain() override {
     // Set if this thread was detached.
     std::unique_ptr<Thread> detached_thread;
 
-    outer_->delegate_->OnMainEntry(outer_);
+    outer_->delegate_->OnMainEntry(outer_.get());
 
     // A SchedulerWorker starts out waiting for work.
     WaitForWork();
 
 #if defined(OS_WIN)
     std::unique_ptr<win::ScopedCOMInitializer> com_initializer;
     if (outer_->backward_compatibility_ ==
         SchedulerBackwardCompatibility::INIT_COM_STA) {
       com_initializer = MakeUnique<win::ScopedCOMInitializer>();
     }
 #endif
 
-    while (!outer_->task_tracker_->IsShutdownComplete() &&
-           !outer_->should_exit_for_testing_.IsSet()) {
+    while (!outer_->ShouldExit()) {
       DCHECK(outer_);
 
 #if defined(OS_MACOSX)
       mac::ScopedNSAutoreleasePool autorelease_pool;
 #endif
 
       UpdateThreadPriority(GetDesiredThreadPriority());
 
       // Get the sequence containing the next task to execute.
-      scoped_refptr<Sequence> sequence = outer_->delegate_->GetWork(outer_);
+      scoped_refptr<Sequence> sequence =
+          outer_->delegate_->GetWork(outer_.get());
       if (!sequence) {
-        if (outer_->delegate_->CanDetach(outer_)) {
-          detached_thread = outer_->Detach();
+        if (outer_->delegate_->CanDetach(outer_.get())) {
+          detached_thread = outer_->DetachThreadObject(DetachNotify::DELEGATE);
           if (detached_thread) {
             outer_ = nullptr;
             DCHECK_EQ(detached_thread.get(), this);
             PlatformThread::Detach(thread_handle_);
             break;
           }
         }
         WaitForWork();
         continue;
       }
@@ skipping 20 matching lines @@
       // sleep if WakeUp() is called while this SchedulerWorker is awake.
       wake_up_event_.Reset();
     }
 
     // If a wake up is pending and we successfully detached, somehow |outer_|
     // was 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_thread || !IsWakeUpPending()) <<
         "This thread was detached and woken up at the same time.";
+
+    // To maintain the condition that the thread always cleans itself up except
+    // in Join(), we go ahead and grab the thread object here. We don't want to

[gab, 2017/02/21 19:01:04]

What makes Join() not hit this code?

[robliao, 2017/02/21 22:26:57]

Join() nulls out |thread_| which means DetachThreadObject() will return nullptr.

[gab, 2017/02/22 18:02:23]

I see and JoinForTesting() also checks for null |thread_| w/ lock before Join()
in case the thread somehow terminated itself already (potentially noticing the
join_called_for_testing event before Join() was invoked in which case the detach
below will be the one reclaiming the Thread object). How about the following:

// Silently make sure the Thread object was reclaimed.
// Cases where this no-ops:
//  - |detached_thread != nullptr|: Thread was explicitly detached above.
//  - JoinForTesting() reclaimed the Thread object.
// Cases where this reclaims the Thread object:
//  - Main loop exited per IsShutdownComplete() or Cleanup().
//  - Main loop exited per JoinForTesting() and this point was reached before
//    JoinForTesting() could even reclaim the Thread object (that's okay: no-op 
//    join).
outer_->DetachThreadObject(DetachNotify::SILENT);

(note: don't even bother with conditional or assigning to variable -- just call
DetachThreadObject(DetachNotify::SILENT) and let it do the right thing)

[robliao, 2017/02/22 20:43:39]

I've enumerated the conditions, but we should keep the check should keep this
check for robustness (it basically costs us nothing and buys us insurance
against future crashes).

It is unexpected that a call to DetachThreadObject would destroy the this
pointer during the execution of ThreadMain.

If the thread detached, the this pointer would get destroyed after the statement
completes. Calling without assignment would result in the same destruction
during execution.

+    // notify the delegate since it's not a detachment.
+    if (!detached_thread)
+      detached_thread = outer_->DetachThreadObject(DetachNotify::SILENT);
   }
 
   void Join() { PlatformThread::Join(thread_handle_); }
 
   void WakeUp() { wake_up_event_.Signal(); }
 
   bool IsWakeUpPending() { return wake_up_event_.IsSignaled(); }
 
  private:
-  Thread(SchedulerWorker* outer)
-      : outer_(outer),
+  Thread(scoped_refptr<SchedulerWorker> outer)
+      : outer_(std::move(outer)),
         wake_up_event_(WaitableEvent::ResetPolicy::MANUAL,
                        WaitableEvent::InitialState::NOT_SIGNALED),
         current_thread_priority_(GetDesiredThreadPriority()) {
     DCHECK(outer_);
   }
 
   void Initialize() {
     constexpr size_t kDefaultStackSize = 0;
     PlatformThread::CreateWithPriority(kDefaultStackSize, this, &thread_handle_,
                                        current_thread_priority_);
@@ skipping 38 matching lines @@
   void UpdateThreadPriority(ThreadPriority desired_thread_priority) {
     if (desired_thread_priority == current_thread_priority_)
       return;
 
     PlatformThread::SetCurrentThreadPriority(desired_thread_priority);
     current_thread_priority_ = desired_thread_priority;
   }
 
   PlatformThreadHandle thread_handle_;
 
-  SchedulerWorker* outer_;
+  scoped_refptr<SchedulerWorker> outer_;
 
   // Event signaled to wake up this thread.
   WaitableEvent wake_up_event_;
 
   // Current priority of this thread. May be different from
   // |outer_->priority_hint_|.
   ThreadPriority current_thread_priority_;
 
   DISALLOW_COPY_AND_ASSIGN(Thread);
 };
 
-std::unique_ptr<SchedulerWorker> SchedulerWorker::Create(
+scoped_refptr<SchedulerWorker> SchedulerWorker::Create(
     ThreadPriority priority_hint,
     std::unique_ptr<Delegate> delegate,
     TaskTracker* task_tracker,
     InitialState initial_state,
     SchedulerBackwardCompatibility backward_compatibility) {
-  auto worker =
-      WrapUnique(new SchedulerWorker(priority_hint, std::move(delegate),
-                                     task_tracker, backward_compatibility));
+  scoped_refptr<SchedulerWorker> worker(
+      new SchedulerWorker(priority_hint, std::move(delegate), task_tracker,
+                          backward_compatibility));
   // Creation happens before any other thread can reference this one, so no
   // synchronization is necessary.
   if (initial_state == SchedulerWorker::InitialState::ALIVE) {
     worker->CreateThread();
     if (!worker->thread_) {
       return nullptr;
     }
   }
 
   return worker;
 }
 
-SchedulerWorker::~SchedulerWorker() {
-  // It is unexpected for |thread_| to be alive and for SchedulerWorker to
-  // destroy since SchedulerWorker owns the delegate needed by |thread_|.
-  // For testing, this generally means JoinForTesting was not called.
-  DCHECK(!thread_);
-}
-
 void SchedulerWorker::WakeUp() {
   AutoSchedulerLock auto_lock(thread_lock_);
 
-  DCHECK(!should_exit_for_testing_.IsSet());
+  DCHECK(!join_called_for_testing_.IsSet());
 
   if (!thread_)
     CreateThreadAssertSynchronized();
 
   if (thread_)
     thread_->WakeUp();
 }
 
 void SchedulerWorker::JoinForTesting() {
-  DCHECK(!should_exit_for_testing_.IsSet());
-  should_exit_for_testing_.Set();
+  DCHECK(!join_called_for_testing_.IsSet());
+  join_called_for_testing_.Set();
 
   std::unique_ptr<Thread> thread;
 
   {
     AutoSchedulerLock auto_lock(thread_lock_);
 
     if (thread_) {
       // Make sure the thread is awake. It will see that
-      // |should_exit_for_testing_| is set and exit shortly after.
+      // |join_called_for_testing_| is set and exit shortly after.
       thread_->WakeUp();
       thread = std::move(thread_);
     }
   }
 
   if (thread)
     thread->Join();
 }
 
 bool SchedulerWorker::ThreadAliveForTesting() const {
   AutoSchedulerLock auto_lock(thread_lock_);
   return !!thread_;
 }
 
+void SchedulerWorker::Cleanup() {
+  // should_exit_ is synchronized with thread_ for writes here so that we can

[gab, 2017/02/21 19:01:04]

|should_exit_|

|thread_|

[robliao, 2017/02/21 22:26:57]

Done.

+  // maintain access |thread_| for wakeup. Otherwise, the thread may take away

[gab, 2017/02/21 19:01:04]

"maintain alive" or just "access" ?

[robliao, 2017/02/21 22:26:57]

Lost a preposition there. Added a "to" -> maintain access to |thread_|

+  // |thread_| for destruction.
+  AutoSchedulerLock auto_lock(thread_lock_);
+  DCHECK(!should_exit_.IsSet());
+  if (thread_) {
+    should_exit_.Set();
+    thread_->WakeUp();
+  }
+}
+
 SchedulerWorker::SchedulerWorker(
     ThreadPriority priority_hint,
     std::unique_ptr<Delegate> delegate,
     TaskTracker* task_tracker,
     SchedulerBackwardCompatibility backward_compatibility)
     : priority_hint_(priority_hint),
       delegate_(std::move(delegate)),
       task_tracker_(task_tracker)
 #if defined(OS_WIN)
       ,
       backward_compatibility_(backward_compatibility)
 #endif
 {
   DCHECK(delegate_);
   DCHECK(task_tracker_);
 }
 
-std::unique_ptr<SchedulerWorker::Thread> SchedulerWorker::Detach() {
+SchedulerWorker::~SchedulerWorker() {
+  // It is unexpected for |thread_| to be alive and for SchedulerWorker to
+  // destroy since SchedulerWorker owns the delegate needed by |thread_|.
+  // For testing, this generally means JoinForTesting was not called.
+  DCHECK(!thread_);
+}
+
+std::unique_ptr<SchedulerWorker::Thread> SchedulerWorker::DetachThreadObject(
+    DetachNotify detach_notify) {
   AutoSchedulerLock auto_lock(thread_lock_);
 
   // Do not detach if the thread is being joined.
   if (!thread_) {
-    DCHECK(should_exit_for_testing_.IsSet());
+    DCHECK(join_called_for_testing_.IsSet());
     return nullptr;
   }
 
   // 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.
   if (thread_->IsWakeUpPending())
     return nullptr;
 
-  // Call OnDetach() within the scope of |thread_lock_| to prevent the delegate
-  // from being used concurrently from an old and a new thread.
-  delegate_->OnDetach();
+  if (detach_notify == DetachNotify::DELEGATE) {
+    // Call OnDetach() within the scope of |thread_lock_| to prevent the
+    // delegate from being used concurrently from an old and a new thread.
+    delegate_->OnDetach();
+  }
 
   return std::move(thread_);
 }
 
 void SchedulerWorker::CreateThread() {
   thread_ = Thread::Create(this);

[gab, 2017/02/21 19:01:04]

Thread::Create(scoped_refptr<SchedulerWorker>(this));

(scoped_refptr isn't currently explicit but ideally would be -- this is what I
was missing in first pass to realize there were two refs around)

[robliao, 2017/02/21 22:26:57]

Went with make_scoped_refptr

 }
 
 void SchedulerWorker::CreateThreadAssertSynchronized() {
   thread_lock_.AssertAcquired();
   CreateThread();
 }
 
+bool SchedulerWorker::ShouldExit() {
+  bool should_exit =
+      task_tracker_->IsShutdownComplete() || join_called_for_testing_.IsSet();

[gab, 2017/02/22 18:02:24]

inline below

[robliao, 2017/02/22 20:43:39]

Done.

+  if (should_exit)
+    return true;
+
+  // We don't need to acquire |thread_lock_| here as the thread will acquire it
+  // on its own to perform changes to thread_.

[gab, 2017/02/22 18:02:24]

|thread_|

[robliao, 2017/02/22 20:43:39]

Removed

+  return should_exit_.IsSet();

[gab, 2017/02/22 18:02:23]

It's unclear why this is checked separately now (it used to be because it
required a lock).

merge as return X || Y || Z;?

[robliao, 2017/02/22 20:43:39]

Done.

+}
+
 }  // namespace internal
 }  // namespace base