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);
 };
 
-// A task runner that runs tasks with the SINGLE_THREADED ExecutionMode.
-class 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) {}
-
-  // 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;
-
-  // Sequence for all Tasks posted through this TaskRunner.
-  const scoped_refptr<Sequence> sequence_ = new Sequence;
-
-  const TaskTraits traits_;
-  SchedulerWorkerPool* 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 TimeDelta& suggested_reclaim_time,
       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);
+  }
+
+  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.
   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 and idle cycle. Workers start out idle.

[fdoray, 2016/07/08 18:58:06]

an* idle cycle

[robliao, 2016/07/08 21:17:03]

Done.

+  bool performed_idle_cycle_ = true;
+
+  subtle::Atomic32 num_single_threaded_runners_ = 0;
+
   const int index_;
 
   DISALLOW_COPY_AND_ASSIGN(SchedulerWorkerDelegateImpl);
 };
 
+// A task runner that runs tasks with the SINGLE_THREADED ExecutionMode.
+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,
+                                  SchedulerWorkerPoolImpl* worker_pool,
+                                  SchedulerWorker* worker)
+      : traits_(traits),
+        worker_pool_(worker_pool),
+        worker_(worker) {
+    DCHECK(worker_pool_);
+    DCHECK(worker_);
+    static_cast<SchedulerWorkerDelegateImpl*>(worker_->delegate())->
+        RegisterSingleThreadTaskRunner();
+  }
+
+  // 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 {
+    static_cast<SchedulerWorkerDelegateImpl*>(worker_->delegate())->
+        UnregisterSingleThreadTaskRunner();
+  }
+
+  // Sequence for all Tasks posted through this TaskRunner.
+  const scoped_refptr<Sequence> sequence_ = new Sequence;
+
+  const TaskTraits traits_;
+  SchedulerWorkerPoolImpl* const worker_pool_;
+  SchedulerWorker* const worker_;
+
+  DISALLOW_COPY_AND_ASSIGN(SchedulerSingleThreadTaskRunner);
+};
+
 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));
   if (worker_pool->Initialize(thread_priority, max_threads,
+                              suggested_reclaim_time,
                               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() {
+  {
+    AutoSchedulerLock auto_lock(join_for_testing_called_lock_);
+    join_for_testing_called_ = true;
+  }
   for (const auto& worker : workers_)
     worker->JoinForTesting();
 
   DCHECK(!join_for_testing_returned_.IsSignaled());
   join_for_testing_returned_.Signal();
 }
 
 scoped_refptr<TaskRunner> SchedulerWorkerPoolImpl::CreateTaskRunnerWithTraits(
     const TaskTraits& traits,
     ExecutionMode execution_mode) {
@@ skipping 100 matching lines @@
     if (worker)
       worker->WakeUp();
     else
       WakeUpOneWorker();
   }
 }
 
 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) {}
 
 SchedulerWorkerPoolImpl::SchedulerWorkerDelegateImpl::
     ~SchedulerWorkerDelegateImpl() = default;
 
 void SchedulerWorkerPoolImpl::SchedulerWorkerDelegateImpl::OnMainEntry(
     SchedulerWorker* worker) {
 #if DCHECK_IS_ON()
   // Wait for |outer_->workers_created_| to avoid traversing
   // |outer_->workers_| while it is being filled by Initialize().
   outer_->workers_created_.Wait();
   DCHECK(ContainsWorker(outer_->workers_, worker));
 #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;
+
   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;
+
   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 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

[fdoray, 2016/07/08 18:58:06]

|num_single_threaded_runners_|

[robliao, 2016/07/08 21:17:03]

Done.

+  // because the next single-threaded task will simply pick up a new physical
+  // thread.
+  bool can_detach = performed_idle_cycle_ &&

[fdoray, 2016/07/08 18:58:07]

const bool

[robliao, 2016/07/08 21:17:03]

Done.

+                    worker != outer_->PeekAtIdleWorkersStack() &&
+                    !subtle::Release_Load(&num_single_threaded_runners_) &&
+                    !outer_->HasJoinedForTesting();

[fdoray, 2016/07/08 18:58:07]

Sadly this is racy:
- Thread A: SchedulerWorker::CanDetach() - returns true.
- Thread B: SchedulerWorkerPoolImpl::JoinForTesting()
- Thread B: SchedulerWorker::JoinForTesting()
- Thread B: Acquires SchedulerWorker::thread_lock_
- Thread A: SchedulerWorker::Detach()
- Thread A: DCHECK(!ShouldExitForTesting()) fails in DCHECK builds.
- Thread A: Tries to acquire SchedulerWorker::thread_lock_.
Dead lock between the join and the acquisition of SchedulerWorker::thread_lock_

I think SchedulerWorker::JoinForTesting() could be rewritten like this:

void SchedulerWorker::JoinForTesting() {
  {
    AutoSchedulerLock auto_lock(should_exit_for_testing_lock_);
    should_exit_for_testing_ = true;
  }
  WakeUp();

  std::unique_ptr<Thread> thread;

  {
    AutoSchedulerLock auto_lock(thread_lock_);
    thread = std::move(thread_);
  }
  
  // Join without holding the lock!
  thread->Join();
}

and SchedulerWorker::Detach() like this:

// Returns nullptr if called during thread->Join(). The thread won't
// detach. It will exit because !outer_->ShouldExitForTesting().
std::unique_ptr<SchedulerWorker::Thread> SchedulerWorker::Detach() {
  DCHECK(!ShouldExitForTesting()) << "Worker was already joined";
  AutoSchedulerLock auto_lock(thread_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 thread_ && thread_->IsWakeUpPending() ? nullptr : std::move(thread_);
}

To remove the requirement that Delegate::CanDetach() must return false when the
thread may be joined.

Maybe we should also add DCHECK(!ShouldExitForTesting()) in WakeUp()?

Let me know if my reasoning makes sense.

[robliao, 2016/07/08 21:17:03]

Yup. This became the first test that flew in the face of this comment in
SchedulerWorker:
// 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.

Because we have no control over the delegate from the test, I added
HasJoinedForTesting because of the race the other way: Thread wakes up after an
idle cycle and attempts to detach after the ShouldExitForTesting check. Also
long as the joins occur before the detach timeout we should be okay.

Alternatively, we can wait until after the detach timeout occurs as well.

I'm not sure we want to go with the suggested code because it walks down a
different problematic road. If we were to make SchedulerWorker robust to Join
and Detach races, now there's no guarantee that the current thread is cleaned up
after Join since it could be in the middle of detaching, which also seems
undesirable for tests that care about that.

It seems that tension comes from having no control on the delegate, so I've
added a disallow control to address this.

Multithreaded code is tricky indeed!

[fdoray, 2016/07/20 14:15:43]

Acknowledged.

+  // CanDetach is part of a SchedulerWorker's idle cycle.
+  performed_idle_cycle_ = true;
+  return can_detach;
 }
 
 SchedulerWorkerPoolImpl::SchedulerWorkerPoolImpl(
     StringPiece name,
     IORestriction io_restriction,
     TaskTracker* task_tracker,
     DelayedTaskManager* delayed_task_manager)
     : name_(name.as_string()),
       io_restriction_(io_restriction),
       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),
+      join_for_testing_called_(false),
 #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 TimeDelta& suggested_reclaim_time,
     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,
+                                 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());
     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
+  // 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::HasJoinedForTesting() {
+  AutoSchedulerLock auto_lock(join_for_testing_called_lock_);
+  return join_for_testing_called_;
+}
+
 }  // namespace internal
 }  // namespace base