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

base/task_scheduler/scheduler_worker_pool_impl.cc

Index: base/task_scheduler/scheduler_worker_pool_impl.cc
diff --git a/base/task_scheduler/scheduler_worker_pool_impl.cc b/base/task_scheduler/scheduler_worker_pool_impl.cc
index 06933eb32db62386195ba3619fd71c03914346fd..954bdfab3b905e2a6653fe5db7ecaacd4247fba3 100644
--- a/base/task_scheduler/scheduler_worker_pool_impl.cc
+++ b/base/task_scheduler/scheduler_worker_pool_impl.cc
@@ -9,6 +9,7 @@
 #include <algorithm>
 #include <utility>
 
+#include "base/atomicops.h"
 #include "base/bind.h"
 #include "base/bind_helpers.h"
 #include "base/lazy_instance.h"
@@ -21,6 +22,7 @@
 #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 {
@@ -43,7 +45,9 @@ class SchedulerParallelTaskRunner : public TaskRunner {
   // 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,
@@ -76,7 +80,9 @@ class SchedulerSequencedTaskRunner : public SequencedTaskRunner {
   // 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,
@@ -121,11 +127,15 @@ class SchedulerSingleThreadTaskRunner : public SingleThreadTaskRunner {
   // TODO(robliao): Find a concrete way to manage the memory of |worker_pool|
   // and |worker|.
   SchedulerSingleThreadTaskRunner(const TaskTraits& traits,
-                                  SchedulerWorkerPool* worker_pool,
+                                  SchedulerWorkerPoolImpl* worker_pool,
                                   SchedulerWorker* worker)
       : traits_(traits),
         worker_pool_(worker_pool),
-        worker_(worker) {}
+        worker_(worker) {
+    DCHECK(worker_pool_);
+    DCHECK(worker_);
+    worker_pool_->RegisterSingleThreadTaskRunner(worker_);
+  }
 
   // SingleThreadTaskRunner:
   bool PostDelayedTask(const tracked_objects::Location& from_here,
@@ -151,13 +161,15 @@ class SchedulerSingleThreadTaskRunner : public SingleThreadTaskRunner {
   }
 
  private:
-  ~SchedulerSingleThreadTaskRunner() override = default;
+  ~SchedulerSingleThreadTaskRunner() override {
+    worker_pool_->UnregisterSingleThreadTaskRunner(worker_);
+  }
 
   // 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);
@@ -187,6 +199,7 @@ class SchedulerWorkerPoolImpl::SchedulerWorkerDelegateImpl
   // the pool name to label the underlying worker threads.
   SchedulerWorkerDelegateImpl(
       SchedulerWorkerPoolImpl* outer,
+      const TimeDelta& suggested_reclaim_time,
       const ReEnqueueSequenceCallback& re_enqueue_sequence_callback,
       const PriorityQueue* shared_priority_queue,
       int index);
@@ -203,8 +216,17 @@ class SchedulerWorkerPoolImpl::SchedulerWorkerDelegateImpl
   TimeDelta GetSleepTimeout() override;
   bool CanDetach(SchedulerWorker* worker) override;
 
+  void RegisterSingleThreadTaskRunner() {
+    subtle::NoBarrier_AtomicIncrement(&num_single_threaded_runners_, 1);
+  }
+
+  void UnregisterSingleThreadTaskRunner() {
+    subtle::NoBarrier_AtomicIncrement(&num_single_threaded_runners_, -1);
+  }
+
  private:
   SchedulerWorkerPoolImpl* outer_;
+  const TimeDelta suggested_reclaim_time_;
   const ReEnqueueSequenceCallback re_enqueue_sequence_callback_;
 
   // Single-threaded PriorityQueue for the worker.
@@ -214,6 +236,8 @@ class SchedulerWorkerPoolImpl::SchedulerWorkerDelegateImpl
   // |single_threaded_priority_queue_|.
   bool last_sequence_is_single_threaded_ = false;
 
+  subtle::Atomic32 num_single_threaded_runners_ = 0;
+
   const int index_;
 
   DISALLOW_COPY_AND_ASSIGN(SchedulerWorkerDelegateImpl);
@@ -231,6 +255,7 @@ std::unique_ptr<SchedulerWorkerPoolImpl> SchedulerWorkerPoolImpl::Create(
     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) {
@@ -238,6 +263,7 @@ std::unique_ptr<SchedulerWorkerPoolImpl> SchedulerWorkerPoolImpl::Create(
       new SchedulerWorkerPoolImpl(name, io_restriction, task_tracker,
                                   delayed_task_manager));
   if (worker_pool->Initialize(thread_priority, max_threads,
+                              suggested_reclaim_time,
                               re_enqueue_sequence_callback)) {
     return worker_pool;
   }
@@ -251,6 +277,10 @@ void SchedulerWorkerPoolImpl::WaitForAllWorkersIdleForTesting() {
 }
 
 void SchedulerWorkerPoolImpl::JoinForTesting() {
+  {
+    AutoSchedulerLock auto_lock(join_for_testing_called_lock_);
+    join_for_testing_called_ = true;
+  }
   for (const auto& worker : workers_)
     worker->JoinForTesting();
 
@@ -368,13 +398,29 @@ void SchedulerWorkerPoolImpl::PostTaskWithSequenceNow(
   }
 }
 
+void SchedulerWorkerPoolImpl::RegisterSingleThreadTaskRunner(
+    SchedulerWorker* worker) {
+  auto delegate =
+      static_cast<SchedulerWorkerDelegateImpl*>(worker->delegate());
+  delegate->RegisterSingleThreadTaskRunner();
+}
+
+void SchedulerWorkerPoolImpl::UnregisterSingleThreadTaskRunner(
+    SchedulerWorker* worker) {
+  auto delegate =
+      static_cast<SchedulerWorkerDelegateImpl*>(worker->delegate());
+  delegate->UnregisterSingleThreadTaskRunner();
+}
+
 SchedulerWorkerPoolImpl::SchedulerWorkerDelegateImpl::
     SchedulerWorkerDelegateImpl(
         SchedulerWorkerPoolImpl* outer,
+        const TimeDelta& suggested_reclaim_time,
         const ReEnqueueSequenceCallback& re_enqueue_sequence_callback,
         const PriorityQueue* shared_priority_queue,
         int index)
     : outer_(outer),
+      suggested_reclaim_time_(suggested_reclaim_time),
       re_enqueue_sequence_callback_(re_enqueue_sequence_callback),
       single_threaded_priority_queue_(shared_priority_queue),
       index_(index) {}
@@ -476,12 +522,14 @@ void SchedulerWorkerPoolImpl::SchedulerWorkerDelegateImpl::
 
 TimeDelta SchedulerWorkerPoolImpl::SchedulerWorkerDelegateImpl::
     GetSleepTimeout() {
-  return TimeDelta::Max();
+  return suggested_reclaim_time_;

[fdoray, 2016/07/04 19:41:32]

From scheduler_worker.cc:

while (...) {
  scoped_refptr<Sequence> sequence = outer_->delegate_->GetWork(outer_);
  if (!sequence) {
    if (outer_->delegate_->CanDetach(outer_)) {
      detached_thread = outer_->Detach();
      ...
    }
    WaitForWork();
    continue;
  }
  ...
}

What prevents the worker from detaching as soon as GetWork() returns nullptr
(i.e. without staying idle for |suggested_reclaim_time_|)?

[robliao, 2016/07/07 17:49:16]

Nice catch! Nothing at the moment. Fixed with an idle cycle bool tracked in
CanDetach()

 }
 
 bool SchedulerWorkerPoolImpl::SchedulerWorkerDelegateImpl::CanDetach(
     SchedulerWorker* worker) {
-  return false;
+  return worker != outer_->PeekAtIdleWorkersStack() &&
+         !subtle::Release_Load(&num_single_threaded_runners_) &&

[fdoray, 2016/07/04 19:41:32]

Add comment explaining why it doesn't matter if |num_single_threaded_runners_|
is incremented right after this (since no task posted to the newly created
single-threaded TaskRunner, there is no need to keep the current "physical"
thread).

[gab, 2016/07/07 15:58:30]

Why do you need Release_Load() here?

[robliao, 2016/07/07 17:49:16]

Done.
Since CanDetach may make decisions based off of surrounding state from the
single threaded task runners (like idle thread accounting), I figured it would
be robust to make sure everything is updated at this point.

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

Hmmm, I don't think this matters, idle thread accounting is done on its own (and
is in fact checked before the ReleaseLoad here anyways). All checks are
self-sufficient so this one is really only about |num_single_threaded_runners|
IMO.

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

In either case, it's the safest thing to do here. We want to make sure all state
from before gets synced up to this point. 

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

I disagree. It's not safer. It's equivalent (plus overhead). CanDetach() doesn't
document that it produces a memory barrier and nothing relies on it doing so.
The only other thing after this call is HasJoinedForTesting() which is
self-sufficient.

Everything in the scheduler is designed to do exactly what it needs to be safe.
Not more, not less.

Adding Release semantics here implicitly encodes a requirement which doesn't
exist. It's like using ThreadTaskRunnerHandle when one merely cares for a
sequence ;-).

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

In light of the atomic ops discussion, any strong opinion to remove the barrier
here?

[gab, 2016/07/20 01:45:21]

From the conclusion we derived, I think the inc/decrements should use Barriers
and the read here should use Acquire_Load(). I'm still not convinced that this
is required when dealing with a single variable (all memory re-ordering examples
I've seen use two variables and even a NoBarrier_Increment has to use RMW --
read-modify-write -- semantics which have an implicit "barrier" for at least
that single variable (but would allow memory re-ordering of other memory regions
with respect to it?), i.e. maybe we could get away with NoBarrier_Increment and
Acquire_Load but I'm not 100% sure anymore...). I am pretty convinced we need
the Acquire_Load() though as the read doesn't have an implicit RMW like the
increments do (don't understand when Release_Load() is appropriate -- C++11
doesn't provide it FWIW).

[robliao, 2016/07/20 19:44:00]

I've my understanding is right, the atomic ops are guaranteed to provide
atomicity in operations. The barriers control visibility.

So IIUC, it's like this

[Atomic Operation]

NoBarrier_AtomicIncrement of the same atomic variable:
Thread A      [In-place Atomic Increment]    [In-place Atomic Increment]
Thread B      [In-place Atomic Increment]    [In-place Atomic Increment]

Result Case 1         0->1                          1->2
Result Case 2               1->2                    2->3
(And more)

The value is atomic (we aren't reading in the middle of another thread writing).
The visibility is up to the processor. 

With Barrier_AtomicIncrement, you impose a total ordering (e.g. mutual
exclusion)

Thread A  [B + In-place Atomic Increment + B]  [B + Barrier In-place Atomic
Increment + B]
Thread B  [B + In-place Atomic Increment + B]  [B + Barrier In-place Atomic
Increment + B]
Result Case 1    A:  0->1                             3->4
                 B:       1->2                   2->3

Result Case 2    A:       1->2                        3->4
                 B:  0->1                        2->3
(...)

Non-atomic increment would separate read-increment-write steps, which for word
sized data types, would lead to similar behavior as NoBarrier_AtomicIncrement.
For sizes larger than a word, we could see intermediate reads and writes.

So that indeed suggests I need a barrier at the increment and decrements.





So as for whether or not to use Acquire_Load and Release_Load, I guess the
question comes down when do I want to sync the world

Acquire [Sync]  Read
Release         Read  [Sync]

Given that this count could be changing any time, practically speaking, it
doesn't seem to matter. If we get a 0, it could be 1 right after the read in
either case. The other thing to consider is the memory around the value. If that
needs to be visible too, then you'll also need a barrier (e.g. a certain
variable A is set to a value iff a variable B is 0).
A Release_Store ensures that setting A and B in concert is visible immediately
after B is set.

In this case of a read, the other values are synchronized, so it doesn't matter
as much if we sync now or later. With this analysis here, I think arguably we
can use NoBarrier_Load since with the increment barriers, the value is visible
everywhere.

[gab, 2016/07/20 20:26:31]

We have different understandings on a few things (inline)
Hmmm separating read-increment-write steps would be wrong even for word sized
data types (the whole point of atomic RMW is to make sure to do so as a single
transaction).
I think you need a barrier so that the result of the increment is "flushed"
(guaranteed to not be reordered after the next sequential read).
I think we do care about syncing before (Acquire), otherwise a thread could see
0 forever. It's not relevant that the count could go up right after check (we
explicitly support that by sending a WakeUp which would result in a sleep no-op
IIRC). What's relevant is that we don't see a value that's outdated beyond the
range we explicitly protect ourselves against.
I don't think we care about that. Nothing depends on being synchronized with the
variable at stake here, adding the Release adds an implicit dependency which we
don't document and that's no good for readability (and a potential burden if
someone starts depending on it because it happens to be there). Every
synchronization in the scheduler is explicit and intentional, let's not add one
for the sake of "why not".
What do you mean by "the other values are synchronized on read"? Which other
values and why and why does it matter?

[robliao, 2016/07/20 22:18:03]

data types (the whole point of atomic RMW is to make sure to do so as a single
transaction).

Indeed, but that's the joy of omitting the barriers. Non-atomic ops look
indistinguishable from atomic ops.
(guaranteed to not be reordered after the next sequential read).
Processors and compilers guarantee that single-threaded behavior is preserved
regardless of the ordering game within a single processor. When multiple
processors (and thus threads) are running, that guarantee goes down to the
memory coherence guarantees (which are quite strong on x86 and weak on arm for
example).
variable at stake here, adding the Release adds an implicit dependency which we
don't document and that's no good for readability (and a potential burden if
someone starts depending on it because it happens to be there). Every
synchronization in the scheduler is explicit and intentional, let's not add one
for the sake of "why not".

That's correct. Since all of the other values are already sync'ed via locks or
are only modified on one thread, we're generally okay.
0 forever. It's not relevant that the count could go up right after check (we
explicitly support that by sending a WakeUp which would result in a sleep no-op
IIRC). What's relevant is that we don't see a value that's outdated beyond the
range we explicitly protect ourselves against.

In the Barrier_AtomicIncrement world, a thread won't always see zero because the
increments are barriered. The memory address in cache is flushed and invalidated
on the other cores.
values and why and why does it matter?
I mean Release_Read. It doesn't matter in this case now that we're using
barriers for the counters.

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

I'm not convinced this is true. The RMW will add an lfence (Acquire) on its read
and an rfence (Release) on its write *for that thread*. That doesn't mean that
other threads will see the value (they need an lfence (Acquire)) to be forced to
fetch it.

In any case, even if NoBarrier was correct here, it's not worth optimizing
CanDetach() as it's running when the thread is idle anyways so might as well add
the lfence (Acquire).

[robliao, 2016/07/21 18:44:23]

Digging into this deeper, the RMW will happen atomically with atomicops (lock
xadd for x86, and a retry loop for arm to make sure the memory written is as
expected). My operating assumption above on visibility incorrectly assumed a
non-atomic RMW.

The fence is to make sure that state is updated in a consistent way. Because
memory ops can be reordered, things will appear just fine on the current thread,
but could appear strange to outside threads. Fencing ensures that memory updates
from a single thread appear in the correct order for other threads.

Using the refcounted object as an example...
When a refcounted object addrefs, the refcount doesn't really matter to an
external thread except for the fact that it is non-zero.

When refcounting down, the comment says to use a barrier to make sure the "state
written before the reference count became zero will be visible to a thread that
has just made the count zero". This comment makes no sense to me because from
the same thread, all memory operations on the same thread are immediately
visible on the same thread (even with reordering, processors work hard to ensure
that single-threaded behaviors still work correctly). 

However, if considering that someone knew they had the last ref (already
problematic, but that's a separate issue), and they set a boolean
fReleasedLastRef, this could potentially be useful to another thread expecting
to see fReleasedLastRef and an object of refcount 0.

[gab, 2016/07/21 21:24:24]

Refcount is different, no one ever wants to just "read" the value (well except
in say AtomicRefCountIsOne() which does us Acquire_Load() as I think we need to
here).

The Barrier on decrement (and that comment) IMO mean that: although the memory
order doesn't matter, whoever reaches 0 (and thus falls last in the relaxed
memory order) should sync its state for *other variables* so that whatever it
deletes (likely action of getting to 0) is the most recent version (and it's
probably guaranteed to see what other threads have done to *other variables*
because their RMW "happened before" in the relaxed memory order per this thread
being the one hitting 0 -- though since it hit 0 last I would assume even a
NoBarrier RMW would force the lfence on the R to sync...).

[robliao, 2016/07/22 16:44:06]

Fenced the load.

+         !outer_->HasJoinedForTesting();
 }
 
 SchedulerWorkerPoolImpl::SchedulerWorkerPoolImpl(
@@ -496,6 +544,7 @@ SchedulerWorkerPoolImpl::SchedulerWorkerPoolImpl(
           idle_workers_stack_lock_.CreateConditionVariable()),
       join_for_testing_returned_(WaitableEvent::ResetPolicy::MANUAL,
                                  WaitableEvent::InitialState::NOT_SIGNALED),
+      join_for_testing_called_(false),
 #if DCHECK_IS_ON()
       workers_created_(WaitableEvent::ResetPolicy::MANUAL,
                        WaitableEvent::InitialState::NOT_SIGNALED),
@@ -509,6 +558,7 @@ SchedulerWorkerPoolImpl::SchedulerWorkerPoolImpl(
 bool SchedulerWorkerPoolImpl::Initialize(
     ThreadPriority thread_priority,
     size_t max_threads,
+    const TimeDelta& suggested_reclaim_time,
     const ReEnqueueSequenceCallback& re_enqueue_sequence_callback) {
   AutoSchedulerLock auto_lock(idle_workers_stack_lock_);
 
@@ -518,10 +568,13 @@ bool SchedulerWorkerPoolImpl::Initialize(
     std::unique_ptr<SchedulerWorker> worker =
         SchedulerWorker::Create(
             thread_priority, WrapUnique(new SchedulerWorkerDelegateImpl(
-                                 this, re_enqueue_sequence_callback,
+                                 this, suggested_reclaim_time,
+                                 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());
@@ -548,18 +601,34 @@ void SchedulerWorkerPoolImpl::WakeUpOneWorker() {
 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
+  // 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))

[fdoray, 2016/07/04 19:41:32]

A detached worker is only re-created when its WakeUp() method is called, right?
WakeUp() can be called from 2 places:

A)
https://cs.chromium.org/chromium/src/base/task_scheduler/scheduler_worker_poo...
A single-threaded task is posted to the worker. The next time that GetWork() is
called, the worker will be removed from the idle stack because work is available
(no other worker can run the single-threaded task).

B)
https://cs.chromium.org/chromium/src/base/task_scheduler/scheduler_worker_poo...
A worker is woken up to run a shared task. Right before this is done, the worker
is popped from the idle stack.

I don't see how detachment makes it possible to try to add the same worker to
the idle stack twice.

[robliao, 2016/07/07 17:49:16]

Prior to this change, GetWork had two preconditions:
1) There may be shared work to be done (The worker has been removed from the
stack via WakeUpOneWorker).
2) There is single-threaded work to be done (The worker will remove itself from
the stack).

With sleep timeouts, this precondition is no longer true since we may have been
woken up by the timeout. This means we'll call GetWork while we're idle, note
that there's no work to be done, and then readd ourselves back onto the idle
workers stack.

[fdoray, 2016/07/07 20:45:56]

You're right. Thanks!

+    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::HasJoinedForTesting() {
+  AutoSchedulerLock auto_lock(join_for_testing_called_lock_);
+  return join_for_testing_called_;
+}
+
 }  // namespace internal
 }  // namespace base