Add Lazy Creation and Thread Detachment Support in the Scheduler Worker Pool

base/task_scheduler/scheduler_worker_pool_impl.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_pool_impl.h"
 
 #include <stddef.h>
 
 #include <algorithm>
 #include <utility>
 
+#include "base/atomicops.h"
 #include "base/bind.h"
 #include "base/bind_helpers.h"
 #include "base/lazy_instance.h"
 #include "base/memory/ptr_util.h"
 #include "base/sequenced_task_runner.h"
 #include "base/single_thread_task_runner.h"
 #include "base/strings/stringprintf.h"
 #include "base/task_scheduler/delayed_task_manager.h"
 #include "base/task_scheduler/task_tracker.h"
 #include "base/threading/platform_thread.h"
 #include "base/threading/thread_local.h"
 #include "base/threading/thread_restrictions.h"
+#include "base/time/time.h"
 
 namespace base {
 namespace internal {
 
 namespace {
 
 // SchedulerWorker that owns the current thread, if any.
 LazyInstance<ThreadLocalPointer<const SchedulerWorker>>::Leaky
     tls_current_worker = LAZY_INSTANCE_INITIALIZER;
 
 // SchedulerWorkerPool that owns the current thread, if any.
 LazyInstance<ThreadLocalPointer<const SchedulerWorkerPool>>::Leaky
     tls_current_worker_pool = LAZY_INSTANCE_INITIALIZER;
 
 // A task runner that runs tasks with the PARALLEL ExecutionMode.
 class SchedulerParallelTaskRunner : public TaskRunner {
  public:
   // Constructs a SchedulerParallelTaskRunner which can be used to post tasks so
   // long as |worker_pool| is alive.
   // TODO(robliao): Find a concrete way to manage |worker_pool|'s memory.
   SchedulerParallelTaskRunner(const TaskTraits& traits,
                               SchedulerWorkerPool* worker_pool)
-      : traits_(traits), worker_pool_(worker_pool) {}
+      : traits_(traits), worker_pool_(worker_pool) {
+    DCHECK(worker_pool_);
+  }
 
   // TaskRunner:
   bool PostDelayedTask(const tracked_objects::Location& from_here,
                        const Closure& closure,
                        TimeDelta delay) override {
     // Post the task as part of a one-off single-task Sequence.
     return worker_pool_->PostTaskWithSequence(
         WrapUnique(new Task(from_here, closure, traits_, delay)),
         make_scoped_refptr(new Sequence), nullptr);
   }
@@ skipping 12 matching lines @@
 };
 
 // A task runner that runs tasks with the SEQUENCED ExecutionMode.
 class SchedulerSequencedTaskRunner : public SequencedTaskRunner {
  public:
   // Constructs a SchedulerSequencedTaskRunner which can be used to post tasks
   // so long as |worker_pool| is alive.
   // TODO(robliao): Find a concrete way to manage |worker_pool|'s memory.
   SchedulerSequencedTaskRunner(const TaskTraits& traits,
                                SchedulerWorkerPool* worker_pool)
-      : traits_(traits), worker_pool_(worker_pool) {}
+      : traits_(traits), worker_pool_(worker_pool) {
+    DCHECK(worker_pool_);
+  }
 
   // SequencedTaskRunner:
   bool PostDelayedTask(const tracked_objects::Location& from_here,
                        const Closure& closure,
                        TimeDelta delay) override {
     std::unique_ptr<Task> task(new Task(from_here, closure, traits_, delay));
     task->sequenced_task_runner_ref = this;
 
     // Post the task as part of |sequence_|.
     return worker_pool_->PostTaskWithSequence(std::move(task), sequence_,
@@ skipping 16 matching lines @@
 
   // Sequence for all Tasks posted through this TaskRunner.
   const scoped_refptr<Sequence> sequence_ = new Sequence;
 
   const TaskTraits traits_;
   SchedulerWorkerPool* const worker_pool_;
 
   DISALLOW_COPY_AND_ASSIGN(SchedulerSequencedTaskRunner);
 };
 
+// Only used in DCHECKs.
+bool ContainsWorker(
+    const std::vector<std::unique_ptr<SchedulerWorker>>& workers,
+    const SchedulerWorker* worker) {
+  auto it = std::find_if(workers.begin(), workers.end(),
+      [worker](const std::unique_ptr<SchedulerWorker>& i) {
+        return i.get() == worker;
+      });
+  return it != workers.end();
+}
+
+}  // namespace
+
 // A task runner that runs tasks with the SINGLE_THREADED ExecutionMode.
-class SchedulerSingleThreadTaskRunner : public SingleThreadTaskRunner {
+class SchedulerWorkerPoolImpl::SchedulerSingleThreadTaskRunner :
+    public SingleThreadTaskRunner {
  public:
   // Constructs a SchedulerSingleThreadTaskRunner which can be used to post
   // tasks so long as |worker_pool| and |worker| are alive.
   // TODO(robliao): Find a concrete way to manage the memory of |worker_pool|
   // and |worker|.
   SchedulerSingleThreadTaskRunner(const TaskTraits& traits,
-                                  SchedulerWorkerPool* worker_pool,
-                                  SchedulerWorker* worker)
-      : traits_(traits),
-        worker_pool_(worker_pool),
-        worker_(worker) {}
+                                  SchedulerWorkerPoolImpl* worker_pool,
+                                  SchedulerWorker* worker);
 
   // SingleThreadTaskRunner:
   bool PostDelayedTask(const tracked_objects::Location& from_here,
                        const Closure& closure,
                        TimeDelta delay) override {
     std::unique_ptr<Task> task(new Task(from_here, closure, traits_, delay));
     task->single_thread_task_runner_ref = this;
 
     // Post the task to be executed by |worker_| as part of |sequence_|.
     return worker_pool_->PostTaskWithSequence(std::move(task), sequence_,
                                               worker_);
   }
 
   bool PostNonNestableDelayedTask(const tracked_objects::Location& from_here,
                                   const Closure& closure,
                                   base::TimeDelta delay) override {
     // Tasks are never nested within the task scheduler.
     return PostDelayedTask(from_here, closure, delay);
   }
 
   bool RunsTasksOnCurrentThread() const override {
     return tls_current_worker.Get().Get() == worker_;
   }
 
  private:
-  ~SchedulerSingleThreadTaskRunner() override = default;
+  ~SchedulerSingleThreadTaskRunner() override;
 
   // Sequence for all Tasks posted through this TaskRunner.
   const scoped_refptr<Sequence> sequence_ = new Sequence;
 
   const TaskTraits traits_;
-  SchedulerWorkerPool* const worker_pool_;
+  SchedulerWorkerPoolImpl* const worker_pool_;
   SchedulerWorker* const worker_;
 
   DISALLOW_COPY_AND_ASSIGN(SchedulerSingleThreadTaskRunner);
 };
 
-// Only used in DCHECKs.
-bool ContainsWorker(
-    const std::vector<std::unique_ptr<SchedulerWorker>>& workers,
-    const SchedulerWorker* worker) {
-  auto it = std::find_if(workers.begin(), workers.end(),
-      [worker](const std::unique_ptr<SchedulerWorker>& i) {
-        return i.get() == worker;
-      });
-  return it != workers.end();
-}
-
-}  // namespace
-
 class SchedulerWorkerPoolImpl::SchedulerWorkerDelegateImpl
     : public SchedulerWorker::Delegate {
  public:
   // |outer| owns the worker for which this delegate is constructed.
   // |re_enqueue_sequence_callback| is invoked when ReEnqueueSequence() is
   // called with a non-single-threaded Sequence. |shared_priority_queue| is a
   // PriorityQueue whose transactions may overlap with the worker's
   // single-threaded PriorityQueue's transactions. |index| will be appended to
   // the pool name to label the underlying worker threads.
   SchedulerWorkerDelegateImpl(
       SchedulerWorkerPoolImpl* outer,
       const ReEnqueueSequenceCallback& re_enqueue_sequence_callback,
       const PriorityQueue* shared_priority_queue,
       int index);
   ~SchedulerWorkerDelegateImpl() override;
 
   PriorityQueue* single_threaded_priority_queue() {
     return &single_threaded_priority_queue_;
   }
 
   // SchedulerWorker::Delegate:
   void OnMainEntry(SchedulerWorker* worker) override;
   scoped_refptr<Sequence> GetWork(SchedulerWorker* worker) override;
   void ReEnqueueSequence(scoped_refptr<Sequence> sequence) override;
   TimeDelta GetSleepTimeout() override;
   bool CanDetach(SchedulerWorker* worker) override;
 
+  void RegisterSingleThreadTaskRunner() {
+    subtle::NoBarrier_AtomicIncrement(&num_single_threaded_runners_, 1);

[gab, 2016/07/13 18:36:31]

Let's wait until Dana/Francois settle the barrier/no-barrier discussion @
https://codereview.chromium.org/2019763002/diff/140001/base/task_scheduler/ta...

[robliao, 2016/07/13 20:19:46]

sgtm. As we are all well aware, atomics are tricky.

+  }
+
+  void UnregisterSingleThreadTaskRunner() {
+    subtle::NoBarrier_AtomicIncrement(&num_single_threaded_runners_, -1);
+  }
+
  private:
   SchedulerWorkerPoolImpl* outer_;
   const ReEnqueueSequenceCallback re_enqueue_sequence_callback_;
 
   // Single-threaded PriorityQueue for the worker.
   PriorityQueue single_threaded_priority_queue_;
 
   // True if the last Sequence returned by GetWork() was extracted from
   // |single_threaded_priority_queue_|.
   bool last_sequence_is_single_threaded_ = false;
 
+  // True if the worker performed an idle cycle. Workers start out idle.
+  bool performed_idle_cycle_ = true;

[gab, 2016/07/13 18:36:31]

is_idle_ ? Then it's true whether it's a new worker or because it became idle?

[robliao, 2016/07/13 20:19:46]

The delegate needs to distinguish between a thread that's idle (GetWork() ==
nullptr) and a thread that's idle after a running an idle cycle (Sleeping and
then waking up again with GetWork() == nullptr).

Tracking is_idle_ is insufficient as that makes detachment happen too early
(detachment is checked after the GetWork() call.

We can't check for detachment after wakeup because we can't distinguish between
wakeups and timeouts (and they're racy).

Checking for an idle cycle is equivalent to checking that we got no work,
attempted to check if we can detach, slept, and then woke up again. If we went
through this cycle once, then we can be assured that we should detach in the
next wake up if there's no work.

[gab, 2016/07/13 21:07:55]

Ah!! Got it :-). Hadn't fully understood the logic here.

Maybe |detach_in_next_cycle_if_no_work_| to be more explicit? Should it actually
start at true?

[robliao, 2016/07/19 22:03:47]

I like performed_idle_cycle_ since it lets us express can_detach in terms of
things that have happened in the SchedulerWorkerDelegateImpl rather than in
terms of itself.

e.g.

can_detach = performed_idle_cycle_ ...

reads better than

can_detach = detach_in_next_cycle_if_no_work_ ...

+
+  subtle::Atomic32 num_single_threaded_runners_ = 0;
+
   const int index_;
 
   DISALLOW_COPY_AND_ASSIGN(SchedulerWorkerDelegateImpl);
 };
 
 SchedulerWorkerPoolImpl::~SchedulerWorkerPoolImpl() {
   // SchedulerWorkerPool should never be deleted in production unless its
   // initialization failed.
   DCHECK(join_for_testing_returned_.IsSignaled() || workers_.empty());
 }
 
 // static
 std::unique_ptr<SchedulerWorkerPoolImpl> SchedulerWorkerPoolImpl::Create(
     StringPiece name,
     ThreadPriority thread_priority,
     size_t max_threads,
     IORestriction io_restriction,
+    const TimeDelta& suggested_reclaim_time,
     const ReEnqueueSequenceCallback& re_enqueue_sequence_callback,
     TaskTracker* task_tracker,
     DelayedTaskManager* delayed_task_manager) {
   std::unique_ptr<SchedulerWorkerPoolImpl> worker_pool(
-      new SchedulerWorkerPoolImpl(name, io_restriction, task_tracker,
-                                  delayed_task_manager));
+      new SchedulerWorkerPoolImpl(name, io_restriction, suggested_reclaim_time,
+                                  task_tracker, delayed_task_manager));
   if (worker_pool->Initialize(thread_priority, max_threads,
                               re_enqueue_sequence_callback)) {
     return worker_pool;
   }
   return nullptr;
 }
 
 void SchedulerWorkerPoolImpl::WaitForAllWorkersIdleForTesting() {
   AutoSchedulerLock auto_lock(idle_workers_stack_lock_);
   while (idle_workers_stack_.Size() < workers_.size())
     idle_workers_stack_cv_for_testing_->Wait();
 }
 
 void SchedulerWorkerPoolImpl::JoinForTesting() {
+  DCHECK(!CanWorkerDetachForTesting() || suggested_reclaim_time_.is_max()) <<
+      "Workers can detach during join.";

[gab, 2016/07/13 18:36:31]

CanWorker...() versus "Workers" (plural versus singular forms). I think it
should be plural everywhere?

[robliao, 2016/07/13 20:19:46]

The act of detaching applies to a single worker, making it singular. A set of
workers can't determine among themselves if detachment is allowed.

   for (const auto& worker : workers_)
     worker->JoinForTesting();
 
   DCHECK(!join_for_testing_returned_.IsSignaled());
   join_for_testing_returned_.Signal();
 }
 
+void SchedulerWorkerPoolImpl::DisallowWorkerDetachmentForTesting() {
+  AutoSchedulerLock auto_lock(worker_detachment_allowed_lock_);
+  worker_detachment_allowed_ = false;
+}
+
 scoped_refptr<TaskRunner> SchedulerWorkerPoolImpl::CreateTaskRunnerWithTraits(
     const TaskTraits& traits,
     ExecutionMode execution_mode) {
   switch (execution_mode) {
     case ExecutionMode::PARALLEL:
       return make_scoped_refptr(new SchedulerParallelTaskRunner(traits, this));
 
     case ExecutionMode::SEQUENCED:
       return make_scoped_refptr(new SchedulerSequencedTaskRunner(traits, this));
 
@@ skipping 90 matching lines @@
                                              sequence_sort_key);
 
     // Wake up a worker to process |sequence|.
     if (worker)
       worker->WakeUp();
     else
       WakeUpOneWorker();
   }
 }
 
+SchedulerWorkerPoolImpl::SchedulerSingleThreadTaskRunner::
+    SchedulerSingleThreadTaskRunner(const TaskTraits& traits,
+                                    SchedulerWorkerPoolImpl* worker_pool,
+                                   SchedulerWorker* worker)

[gab, 2016/07/13 18:36:31]

missing space: git cl format?

[robliao, 2016/07/13 20:19:46]

Done.

+    : traits_(traits),
+      worker_pool_(worker_pool),
+      worker_(worker) {
+  DCHECK(worker_pool_);
+  DCHECK(worker_);
+  static_cast<SchedulerWorkerDelegateImpl*>(worker_->delegate())->
+      RegisterSingleThreadTaskRunner();
+}
+
+SchedulerWorkerPoolImpl::SchedulerSingleThreadTaskRunner::
+    ~SchedulerSingleThreadTaskRunner() {
+  static_cast<SchedulerWorkerDelegateImpl*>(worker_->delegate())->
+      UnregisterSingleThreadTaskRunner();
+}
+
 SchedulerWorkerPoolImpl::SchedulerWorkerDelegateImpl::
     SchedulerWorkerDelegateImpl(
         SchedulerWorkerPoolImpl* outer,
         const ReEnqueueSequenceCallback& re_enqueue_sequence_callback,
         const PriorityQueue* shared_priority_queue,
         int index)
     : outer_(outer),
       re_enqueue_sequence_callback_(re_enqueue_sequence_callback),
       single_threaded_priority_queue_(shared_priority_queue),
       index_(index) {}
@@ skipping 11 matching lines @@
 #endif
 
   PlatformThread::SetName(
       StringPrintf("%sWorker%d", outer_->name_.c_str(), index_));
 
   DCHECK(!tls_current_worker.Get().Get());
   DCHECK(!tls_current_worker_pool.Get().Get());
   tls_current_worker.Get().Set(worker);
   tls_current_worker_pool.Get().Set(outer_);
 
+  // Workers start out idle, so this counts as an idle cycle.
+  performed_idle_cycle_ = true;

[gab, 2016/07/13 18:36:31]

This is redundant per the value being initialized to true.

[robliao, 2016/07/13 20:19:46]

The delegate is reused across thread instances, so we can have multiple
MainEntry calls per delegate. The state needs to be reset here.

+
   ThreadRestrictions::SetIOAllowed(outer_->io_restriction_ ==
                                    IORestriction::ALLOWED);
 }
 
 scoped_refptr<Sequence>
 SchedulerWorkerPoolImpl::SchedulerWorkerDelegateImpl::GetWork(
     SchedulerWorker* worker) {
   DCHECK(ContainsWorker(outer_->workers_, worker));
 
   scoped_refptr<Sequence> sequence;
@@ skipping 36 matching lines @@
       sequence = shared_transaction->PopSequence();
       last_sequence_is_single_threaded_ = false;
     } else {
       DCHECK(!single_threaded_transaction->IsEmpty());
       sequence = single_threaded_transaction->PopSequence();
       last_sequence_is_single_threaded_ = true;
     }
   }
   DCHECK(sequence);
 
+  // We're doing work, so this is not an idle cycle.
+  performed_idle_cycle_ = false;

[gab, 2016/07/13 18:36:31]

is_idle_ = false;

with no comments is sufficient IMO.

[robliao, 2016/07/13 20:19:46]

sgtm. Removed.

+
   outer_->RemoveFromIdleWorkersStack(worker);
   return sequence;
 }
 
 void SchedulerWorkerPoolImpl::SchedulerWorkerDelegateImpl::
     ReEnqueueSequence(scoped_refptr<Sequence> sequence) {
   if (last_sequence_is_single_threaded_) {
     // A single-threaded Sequence is always re-enqueued in the single-threaded
     // PriorityQueue from which it was extracted.
     const SequenceSortKey sequence_sort_key = sequence->GetSortKey();
     single_threaded_priority_queue_.BeginTransaction()->Push(
         std::move(sequence), sequence_sort_key);
   } else {
     // |re_enqueue_sequence_callback_| will determine in which PriorityQueue
     // |sequence| must be enqueued.
     re_enqueue_sequence_callback_.Run(std::move(sequence));
   }
 }
 
 TimeDelta SchedulerWorkerPoolImpl::SchedulerWorkerDelegateImpl::
     GetSleepTimeout() {
-  return TimeDelta::Max();
+  return outer_->suggested_reclaim_time_;
 }
 
 bool SchedulerWorkerPoolImpl::SchedulerWorkerDelegateImpl::CanDetach(
     SchedulerWorker* worker) {
-  return false;
+  // It's not an issue if |num_single_threaded_runners_| is incremented after
+  // this because the next single-threaded task will simply pick up a new
+  // physical thread.
+  const bool can_detach =
+      performed_idle_cycle_ &&
+      worker != outer_->PeekAtIdleWorkersStack() &&
+      !subtle::Release_Load(&num_single_threaded_runners_) &&
+      outer_->CanWorkerDetachForTesting();
+  // CanDetach is part of a SchedulerWorker's idle cycle.
+  performed_idle_cycle_ = true;

[gab, 2016/07/13 18:36:31]

Shouldn't is_idle_ instead be set when about to return no work from GetWork()?
It seems weird to me to check |is_idle_| and then set it to true after?

[robliao, 2016/07/13 20:19:46]

See is_idle_ comment.

[gab, 2016/07/13 21:07:55]

Perhaps expand comment above here to explain the detachment dance logic you just
explained to me in that other comment? (your explanation there was very
compelling but not captured in code comments AFAICT)

[robliao, 2016/07/19 22:03:47]

Done.

+  return can_detach;
 }
 
 SchedulerWorkerPoolImpl::SchedulerWorkerPoolImpl(
     StringPiece name,
     IORestriction io_restriction,
+    const TimeDelta& suggested_reclaim_time,
     TaskTracker* task_tracker,
     DelayedTaskManager* delayed_task_manager)
     : name_(name.as_string()),
       io_restriction_(io_restriction),
+      suggested_reclaim_time_(suggested_reclaim_time),
       idle_workers_stack_lock_(shared_priority_queue_.container_lock()),
       idle_workers_stack_cv_for_testing_(
           idle_workers_stack_lock_.CreateConditionVariable()),
       join_for_testing_returned_(WaitableEvent::ResetPolicy::MANUAL,
                                  WaitableEvent::InitialState::NOT_SIGNALED),
+      worker_detachment_allowed_(true),

[gab, 2016/07/13 18:36:31]

If keeping the bool (ref. WaitableEvent comment), initialize it at declaration
site.

[robliao, 2016/07/13 20:19:46]

See earlier comment.

 #if DCHECK_IS_ON()
       workers_created_(WaitableEvent::ResetPolicy::MANUAL,
                        WaitableEvent::InitialState::NOT_SIGNALED),
 #endif
       task_tracker_(task_tracker),
       delayed_task_manager_(delayed_task_manager) {
   DCHECK(task_tracker_);
   DCHECK(delayed_task_manager_);
 }
 
 bool SchedulerWorkerPoolImpl::Initialize(
     ThreadPriority thread_priority,
     size_t max_threads,
     const ReEnqueueSequenceCallback& re_enqueue_sequence_callback) {
   AutoSchedulerLock auto_lock(idle_workers_stack_lock_);
 
   DCHECK(workers_.empty());
 
   for (size_t i = 0; i < max_threads; ++i) {
     std::unique_ptr<SchedulerWorker> worker =
         SchedulerWorker::Create(
             thread_priority, WrapUnique(new SchedulerWorkerDelegateImpl(
                                  this, re_enqueue_sequence_callback,
                                  &shared_priority_queue_, static_cast<int>(i))),
             task_tracker_,
-            SchedulerWorker::InitialState::ALIVE);
+            i == 0
+                ? SchedulerWorker::InitialState::ALIVE
+                : SchedulerWorker::InitialState::DETACHED);
     if (!worker)
       break;
     idle_workers_stack_.Push(worker.get());
     workers_.push_back(std::move(worker));
   }
 
 #if DCHECK_IS_ON()
   workers_created_.Signal();
 #endif
 
   return !workers_.empty();
 }
 
 void SchedulerWorkerPoolImpl::WakeUpOneWorker() {
   SchedulerWorker* worker;
   {
     AutoSchedulerLock auto_lock(idle_workers_stack_lock_);
     worker = idle_workers_stack_.Pop();
   }
   if (worker)
     worker->WakeUp();
 }
 
 void SchedulerWorkerPoolImpl::AddToIdleWorkersStack(
     SchedulerWorker* worker) {
   AutoSchedulerLock auto_lock(idle_workers_stack_lock_);
-  idle_workers_stack_.Push(worker);
+  // Detachment may cause multiple attempts to add because the delegate cannot

[gab, 2016/07/13 18:36:31]

Shouldn't a thread that wakes up after it's sleep timeout for detachment detach
without bothering to GetWork()? I guess we need to check GetWork() in the racy
event that the thread was actually woken for work but happens to only get a CPU
slot after its detach deadline? And hence we need this? (just making sure I
follow :-)!)

[robliao, 2016/07/13 20:19:46]

We can't distinguish between sleep timeouts and actual wakeups. They look the
same from the thread's perspective. Add to that the race, and we have to check
GetWork in all cases.

+  // determine who woke it up. As a result, when it wakes up, it may conclude
+  // there's no work to be done and attempt to add itself to the idle stack
+  // again.
+  if (!idle_workers_stack_.Contains(worker))
+    idle_workers_stack_.Push(worker);
+
   DCHECK_LE(idle_workers_stack_.Size(), workers_.size());
 
   if (idle_workers_stack_.Size() == workers_.size())
     idle_workers_stack_cv_for_testing_->Broadcast();
 }
 
+const SchedulerWorker* SchedulerWorkerPoolImpl::PeekAtIdleWorkersStack() const {
+  AutoSchedulerLock auto_lock(idle_workers_stack_lock_);
+  return idle_workers_stack_.Peek();
+}
+
 void SchedulerWorkerPoolImpl::RemoveFromIdleWorkersStack(
     SchedulerWorker* worker) {
   AutoSchedulerLock auto_lock(idle_workers_stack_lock_);
   idle_workers_stack_.Remove(worker);
 }
 
+bool SchedulerWorkerPoolImpl::CanWorkerDetachForTesting() {
+  AutoSchedulerLock auto_lock(worker_detachment_allowed_lock_);
+  return worker_detachment_allowed_;
+}
+
 }  // namespace internal
 }  // namespace base