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

base/task_scheduler/scheduler_worker_pool_impl_unittest.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 <memory>
 #include <unordered_set>
 #include <vector>
 
+#include "base/atomicops.h"
 #include "base/bind.h"
 #include "base/bind_helpers.h"
 #include "base/callback.h"
 #include "base/macros.h"
 #include "base/memory/ptr_util.h"
 #include "base/memory/ref_counted.h"
 #include "base/synchronization/condition_variable.h"
 #include "base/synchronization/lock.h"
 #include "base/synchronization/waitable_event.h"
 #include "base/task_runner.h"
 #include "base/task_scheduler/delayed_task_manager.h"
 #include "base/task_scheduler/sequence.h"
 #include "base/task_scheduler/sequence_sort_key.h"
 #include "base/task_scheduler/task_tracker.h"
 #include "base/task_scheduler/test_task_factory.h"
 #include "base/task_scheduler/test_utils.h"
 #include "base/threading/platform_thread.h"
 #include "base/threading/simple_thread.h"
+#include "base/threading/thread_local_storage.h"
 #include "base/threading/thread_restrictions.h"
+#include "base/time/time.h"
 #include "testing/gtest/include/gtest/gtest.h"
 
 namespace base {
 namespace internal {
 namespace {
 
 const size_t kNumWorkersInWorkerPool = 4;
 const size_t kNumThreadsPostingTasks = 4;
 const size_t kNumTasksPostedPerThread = 150;
+const TimeDelta kSuggestedReclaimTime = TimeDelta::FromMilliseconds(10);
 
 using IORestriction = SchedulerWorkerPoolImpl::IORestriction;
 
 class TestDelayedTaskManager : public DelayedTaskManager {
  public:
   TestDelayedTaskManager() : DelayedTaskManager(Bind(&DoNothing)) {}
 
   void SetCurrentTime(TimeTicks now) { now_ = now; }
 
   // DelayedTaskManager:
   TimeTicks Now() const override { return now_; }
 
  private:
   TimeTicks now_ = TimeTicks::Now();
 
   DISALLOW_COPY_AND_ASSIGN(TestDelayedTaskManager);
 };
 
 class TaskSchedulerWorkerPoolImplTest
     : public testing::TestWithParam<ExecutionMode> {
  protected:
   TaskSchedulerWorkerPoolImplTest() = default;
 
   void SetUp() override {
-    worker_pool_ = SchedulerWorkerPoolImpl::Create(
-        "TestWorkerPoolWithFileIO", ThreadPriority::NORMAL,
-        kNumWorkersInWorkerPool, IORestriction::ALLOWED,
-        Bind(&TaskSchedulerWorkerPoolImplTest::ReEnqueueSequenceCallback,
-             Unretained(this)),
-        &task_tracker_, &delayed_task_manager_);
-    ASSERT_TRUE(worker_pool_);
+    InitializeWorkerPool();
+    worker_pool_->DisallowWorkerDetachmentForTesting();
   }
 
   void TearDown() override {
     worker_pool_->WaitForAllWorkersIdleForTesting();
     worker_pool_->JoinForTesting();
   }
 
+  void InitializeWorkerPool() {
+    worker_pool_ = SchedulerWorkerPoolImpl::Create(
+        "TestWorkerPoolWithFileIO", ThreadPriority::NORMAL,
+        kNumWorkersInWorkerPool, IORestriction::ALLOWED,
+        kSuggestedReclaimTime,
+        Bind(&TaskSchedulerWorkerPoolImplTest::ReEnqueueSequenceCallback,
+             Unretained(this)),
+        &task_tracker_, &delayed_task_manager_);
+    ASSERT_TRUE(worker_pool_);
+  }
+
   std::unique_ptr<SchedulerWorkerPoolImpl> worker_pool_;
 
   TaskTracker task_tracker_;
   TestDelayedTaskManager delayed_task_manager_;
 
  private:
   void ReEnqueueSequenceCallback(scoped_refptr<Sequence> sequence) {
     // In production code, this callback would be implemented by the
     // TaskScheduler which would first determine which PriorityQueue the
     // sequence must be re-enqueued.
@@ skipping 268 matching lines @@
 };
 
 }  // namespace
 
 TEST_P(TaskSchedulerWorkerPoolImplIORestrictionTest, IORestriction) {
   TaskTracker task_tracker;
   DelayedTaskManager delayed_task_manager(Bind(&DoNothing));
 
   auto worker_pool = SchedulerWorkerPoolImpl::Create(
       "TestWorkerPoolWithParam", ThreadPriority::NORMAL, 1U, GetParam(),
-      Bind(&NotReachedReEnqueueSequenceCallback), &task_tracker,
-      &delayed_task_manager);
+      TimeDelta::Max(), Bind(&NotReachedReEnqueueSequenceCallback),
+      &task_tracker, &delayed_task_manager);
   ASSERT_TRUE(worker_pool);
+  worker_pool->DisallowWorkerDetachmentForTesting();
 
   WaitableEvent task_ran(WaitableEvent::ResetPolicy::MANUAL,
                          WaitableEvent::InitialState::NOT_SIGNALED);
   worker_pool->CreateTaskRunnerWithTraits(TaskTraits(), ExecutionMode::PARALLEL)
       ->PostTask(FROM_HERE, Bind(&ExpectIORestriction, GetParam(), &task_ran));
   task_ran.Wait();
 
   worker_pool->JoinForTesting();
 }
 
 INSTANTIATE_TEST_CASE_P(IOAllowed,
                         TaskSchedulerWorkerPoolImplIORestrictionTest,
                         ::testing::Values(IORestriction::ALLOWED));
 INSTANTIATE_TEST_CASE_P(IODisallowed,
                         TaskSchedulerWorkerPoolImplIORestrictionTest,
                         ::testing::Values(IORestriction::DISALLOWED));
 
+namespace {
+
+constexpr size_t kMagicTlsValue = 42;
+
+class TaskSchedulerWorkerPoolSingleThreadedTest
+    : public TaskSchedulerWorkerPoolImplTest {
+ public:
+  void SetTlsValue() {
+    slot_.Set(reinterpret_cast<void*>(kMagicTlsValue));
+  }
+
+  void CheckTlsValue() {
+    EXPECT_EQ(kMagicTlsValue, reinterpret_cast<size_t>(slot_.Get()));

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

Instead of using TLS to do an "is this the right thread" check, I think it's
cleaner to use ThreadCheckerImpl.

To set it you can do:

// Re-attach |thread_checker_| to the current thread.
thread_checker_.DetachFromThread();
thread_checker.CalledOnValidThread();

And then simply use CalledOnValidThread() in the check phase.

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

Nice. I forgot about ThreadCheckerImpl. Done.

+  }
+
+ protected:
+  void SetUp() override {
+    InitializeWorkerPool();
+  }
+
+  TaskSchedulerWorkerPoolSingleThreadedTest() = default;
+
+ private:
+  ThreadLocalStorage::Slot slot_;
+
+  DISALLOW_COPY_AND_ASSIGN(TaskSchedulerWorkerPoolSingleThreadedTest);
+};
+
+}  // namespace
+
+// Verify that thread resources for a single thread remain.
+TEST_F(TaskSchedulerWorkerPoolSingleThreadedTest, SingleThreadTask) {
+  auto single_thread_task_runner =
+      worker_pool_->CreateTaskRunnerWithTraits(
+          TaskTraits().
+              WithShutdownBehavior(TaskShutdownBehavior::BLOCK_SHUTDOWN),
+          ExecutionMode::SINGLE_THREADED);
+  single_thread_task_runner->PostTask(
+      FROM_HERE,
+      Bind(&TaskSchedulerWorkerPoolSingleThreadedTest::SetTlsValue,
+           Unretained(this)));
+  WaitableEvent task_waiter(WaitableEvent::ResetPolicy::MANUAL,
+                            WaitableEvent::InitialState::NOT_SIGNALED);
+  single_thread_task_runner->PostTask(
+      FROM_HERE, Bind(&WaitableEvent::Signal, Unretained(&task_waiter)));
+  task_waiter.Wait();
+  task_waiter.Reset();

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

Make the event AUTOMATIC instead of doing Wait()+Reset()

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

I thought we were preferring Manual events for tests?

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

I don't think there's a preference. I tend to use MANUAL unless I care about it
being reset for re-use which you do here.

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

Cool. I prefer automatic here as well. Done.

+  worker_pool_->WaitForAllWorkersIdleForTesting();
+
+  // Give the worker pool a chance to reclaim its threads.

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

It was never forced to create more than one thread above so I don't understand
what this means? IMO to test this you need N blocking tasks (N-1 of which are
not single-threaded). Release the single-threaded one first, then the others.
And confirm that despite that the single-threaded task can still run on the same
thread (i.e. it wasn't among the terminated threads).

Would be nice to verify actual thread count in process too to confirm that
threads actually were created and wound down (confirming that things happened
and single-threaded bound thread was protected).

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

The goal of this test is to verify that the worker pool does not reclaim the
thread with single threaded work after an idle cycle. This test fails if the
thread was allowed to reclaim (e.g. CanDetach always returns true). From the
caller's standpoint, we don't know how many threads were created. All we know is
that stuff can happen during the suggested reclaim time, so we want for that
long and then some to ensure things have settled down.

I wanted to avoid checking thread counts since I don't think there are
guarantees for stable thread counts in a testing environment. 

+  PlatformThread::Sleep(
+      kSuggestedReclaimTime + TimeDelta::FromMilliseconds(200));
+
+  worker_pool_->DisallowWorkerDetachmentForTesting();
+
+  single_thread_task_runner->PostTask(
+      FROM_HERE,
+      Bind(&TaskSchedulerWorkerPoolSingleThreadedTest::CheckTlsValue,
+           Unretained(this)));
+  single_thread_task_runner->PostTask(
+      FROM_HERE, Bind(&WaitableEvent::Signal, Unretained(&task_waiter)));
+  task_waiter.Wait();
+}
+
+namespace {
+
+class TaskSchedulerWorkerPoolCheckTlsReuse
+    : public TaskSchedulerWorkerPoolImplTest {
+ public:
+  void SetTlsValueAndWait() {
+    slot_.Set(reinterpret_cast<void*>(kMagicTlsValue));
+    waiter_.Wait();
+  }
+
+  void CountZeroTlsValuesAndWait(WaitableEvent* count_waiter) {
+    if (!slot_.Get())
+      subtle::NoBarrier_AtomicIncrement(&zero_tls_values_, 1);
+
+    count_waiter->Signal();
+    waiter_.Wait();
+  }
+
+ protected:
+  TaskSchedulerWorkerPoolCheckTlsReuse() :
+      waiter_(WaitableEvent::ResetPolicy::MANUAL,
+              WaitableEvent::InitialState::NOT_SIGNALED) {}
+
+  void SetUp() override {
+    InitializeWorkerPool();
+  }
+
+  subtle::Atomic32 zero_tls_values_ = 0;
+
+  WaitableEvent waiter_;
+
+ private:
+  ThreadLocalStorage::Slot slot_;
+
+  DISALLOW_COPY_AND_ASSIGN(TaskSchedulerWorkerPoolCheckTlsReuse);
+};
+
+}  // namespace
+
+// Checks that at least one thread has detached by checking the TLS.
+TEST_F(TaskSchedulerWorkerPoolCheckTlsReuse, CheckDetachedThreads) {
+  // Saturate the threads and mark each thread with a magic TLS value.
+  std::vector<std::unique_ptr<test::TestTaskFactory>> factories;
+  for (size_t i = 0; i < kNumWorkersInWorkerPool; ++i) {
+    factories.push_back(WrapUnique(new test::TestTaskFactory(
+        worker_pool_->CreateTaskRunnerWithTraits(
+            TaskTraits(), ExecutionMode::PARALLEL),
+        ExecutionMode::PARALLEL)));
+    ASSERT_TRUE(factories.back()->PostTask(
+        PostNestedTask::NO,
+        Bind(&TaskSchedulerWorkerPoolCheckTlsReuse::SetTlsValueAndWait,
+             Unretained(this))));
+    factories.back()->WaitForAllTasksToRun();
+  }
+
+  // Release tasks waiting on |waiter_|.
+  waiter_.Signal();
+  worker_pool_->WaitForAllWorkersIdleForTesting();
+
+  // All threads should be done running by now, so reset for the next phase.
+  waiter_.Reset();
+
+  // Give the worker pool a chance to detach its threads.
+  PlatformThread::Sleep(
+      kSuggestedReclaimTime + TimeDelta::FromMilliseconds(200));
+
+  worker_pool_->DisallowWorkerDetachmentForTesting();
+
+  // Saturate and count the threads that do not have the magic TLS value. If the
+  // value is not there, that means we're at a new thread.
+  std::vector<std::unique_ptr<WaitableEvent>> count_waiters;
+  for (auto& factory : factories) {
+    count_waiters.push_back(WrapUnique(new WaitableEvent(
+        WaitableEvent::ResetPolicy::MANUAL,
+        WaitableEvent::InitialState::NOT_SIGNALED)));
+    ASSERT_TRUE(factory->PostTask(
+          PostNestedTask::NO,
+          Bind(&TaskSchedulerWorkerPoolCheckTlsReuse::CountZeroTlsValuesAndWait,

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

Instead of counting lack of TLS values here I think a cleaner approach would be
to build a std::set of PlatformThreadRef in the first phase and check against it
in the second phase (effectively ThreadCheckerImpl but pooled).

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

Wouldn't using the TLS be basically the same thing except with less bookkeeping?
The test doesn't actually care about the thread instances themselves, but rather
that the environment is clean when it entered. TLS's are guaranteed by the OS to
be 0/null upon entry into a new thread.

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

Ah ok you're right, the TLS effectively acts as a "one of these threads" flag
without having to bookkeep threads themselves.

+               Unretained(this),
+               count_waiters.back().get())));
+    factory->WaitForAllTasksToRun();
+  }
+
+  // Wait for all counters to complete.
+  for (auto& count_waiter : count_waiters)
+    count_waiter->Wait();
+
+  EXPECT_GT(zero_tls_values_, 0);
+
+  // Release tasks waiting on |waiter_|.
+  waiter_.Signal();
+}
+
 }  // namespace internal
 }  // namespace base