DESIGN DISCUSSION ONLY Task Scheduler Single Thread Task Runner Manager for Dedicated Threads per S…

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>
 
@@ skipping 80 matching lines @@
       // 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 SchedulerWorker 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 SchedulerWorker is awake.
       wake_up_event_.Reset();
     }

[fdoray, 2017/02/08 17:59:01]

There is a leak if someone does this:

TEST(...) {
  base::TaskScheduler::CreateAndSetSimpleTaskScheduler(...); 
  auto task_runner = base::CreateSingleThreadTaskRunnerWithTraits();
  task_runner->PostTask(...);  

  // Drop the last reference to |task_runner|.
  // Ownership of the SchedulerWorker is transferred to
  // SchedulerWorker::Thread. SchedulerWorker::Thread is
  // owned by SchedulerWorker.
  task_runner = nullptr;

  // This allows the SchedulerWorker::Thread::ThreadMain
  // to return without ownership of SchedulerWorker::Thread
  // being transferred to |detached_thread|.
  base::TaskScheduler::GetInstance()->Shutdown();

  base::TaskScheduler::GetInstance()->JoinForTesting();
  base::TaskScheduler::SetInstance(nullptr);

  // SchedulerWorker and SchedulerWorker::Thread are leaked.
}

You could remove lines 71-72 and add this here:

if (!detached_thread) {
  // When |owning_worker_| is non-nullptr, |this| owns
  // |owning_worker_| and |owning_worker_| owns this.
  // Break the cycle by taking ownership of |this|.
  detached_thread = outer_->DetachEvenIfWakeUpPending();
}

if (detached_thread) {
  DCHECK_EQ(detached_thread.get(), this);
  PlatformThread::Detach(thread_handle_);
}

And in JoinForTesting(), you could move should_exit_for_testing_.Set(); inside
the lock to prevent a detach during a join.

[robliao, 2017/02/11 02:13:34]

Thanks for pointing this out. This is covered by a new unit test in the real CL.
TEST(TaskSchedulerWorkerTest,
     WorkerDetachesAndSelfDestroysDuringShutdown) 

 
     // 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.";
   }
 
+  void TakeOwnershipOfSchedulerWorker(std::unique_ptr<SchedulerWorker> worker) {
+    // If this check fails, we'll crash on a random thread instead.
+    CHECK(outer_ == worker.get());

[fdoray, 2017/02/08 17:59:01]

DCHECK_EQ

[robliao, 2017/02/10 00:04:04]

This CHECK is intentional. I'd prefer to crash right here instead of somewhere
mysterious.

+    owning_worker_ = std::move(worker);

[fdoray, 2017/02/08 17:59:01]

WakeUp();

Otherwise, the thread may sleep forever and the SchedulerWorker may never be
deleted. You will need to remove
https://cs.chromium.org/chromium/src/base/task_scheduler/scheduler_worker.cc?....

[robliao, 2017/02/11 02:13:34]

Done.

+  }
+
   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),
         wake_up_event_(WaitableEvent::ResetPolicy::MANUAL,
@@ skipping 56 matching lines @@
 
   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_;
 
+  // Holds the owning worker for destruction at the end of detachment.
+  std::unique_ptr<SchedulerWorker> owning_worker_;
+
   DISALLOW_COPY_AND_ASSIGN(Thread);
 };
 
 std::unique_ptr<SchedulerWorker> SchedulerWorker::Create(
     ThreadPriority priority_hint,
     std::unique_ptr<Delegate> delegate,
     TaskTracker* task_tracker,
     InitialState initial_state,
     SchedulerBackwardCompatibility backward_compatibility) {
   auto worker =
@@ skipping 11 matching lines @@
   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_);
 }
 
+// static
+void SchedulerWorker::DestroyAfterDetachment(
+    std::unique_ptr<SchedulerWorker> worker) {
+  DCHECK(worker);
+  AutoSchedulerLock auto_lock(worker->thread_lock_);
+  if (worker->thread_) {
+    SchedulerWorker* raw_worker_pointer = worker.get();
+    raw_worker_pointer->thread_->TakeOwnershipOfSchedulerWorker(
+        std::move(worker));
+  }
+}
+
 void SchedulerWorker::WakeUp() {
   AutoSchedulerLock auto_lock(thread_lock_);
 
   DCHECK(!should_exit_for_testing_.IsSet());
 
   if (!thread_)
     CreateThreadAssertSynchronized();
 
   if (thread_)
     thread_->WakeUp();
@@ skipping 68 matching lines @@
   thread_ = Thread::Create(this);
 }
 
 void SchedulerWorker::CreateThreadAssertSynchronized() {
   thread_lock_.AssertAcquired();
   CreateThread();
 }
 
 }  // namespace internal
 }  // namespace base