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/scheduler_worker_pool_params.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_checker_impl.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);

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

"kSuggestedReclaimTimeForDetachTests" or something (otherwise I find it
unexpected that it's not used directly in InitializeWorkerPool())

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

constexpr

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

Done.

 
 using IORestriction = SchedulerWorkerPoolParams::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(
-        SchedulerWorkerPoolParams("TestWorkerPoolWithFileIO",
-                                  ThreadPriority::NORMAL,
-                                  IORestriction::ALLOWED,
-                                  kNumWorkersInWorkerPool),
-        Bind(&TaskSchedulerWorkerPoolImplTest::ReEnqueueSequenceCallback,
-             Unretained(this)),
-        &task_tracker_, &delayed_task_manager_);
-    ASSERT_TRUE(worker_pool_);
+    InitializeWorkerPool(TimeDelta::Max());
+    worker_pool_->DisallowWorkerDetachmentForTesting();

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

This is not required given that suggested reclaim time is TimeDelta::Max(),
right?

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

Yep. Done.

   }
 
   void TearDown() override {
     worker_pool_->WaitForAllWorkersIdleForTesting();
     worker_pool_->JoinForTesting();
   }
 
+  void InitializeWorkerPool(const TimeDelta& suggested_reclaim_time) {
+    worker_pool_ = SchedulerWorkerPoolImpl::Create(
+        SchedulerWorkerPoolParams("TestWorkerPoolWithFileIO",
+                                  ThreadPriority::NORMAL,
+                                  IORestriction::ALLOWED,
+                                  kNumWorkersInWorkerPool,
+                                  suggested_reclaim_time),
+        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(
       SchedulerWorkerPoolParams("TestWorkerPoolWithParam",
-                                ThreadPriority::NORMAL, GetParam(), 1U),
+                                ThreadPriority::NORMAL, GetParam(), 1U,
+                                TimeDelta::Max()),
       Bind(&NotReachedReEnqueueSequenceCallback), &task_tracker,
       &delayed_task_manager);
   ASSERT_TRUE(worker_pool);
+  worker_pool->DisallowWorkerDetachmentForTesting();

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

This is not required given that suggested reclaim time is TimeDelta::Max(),
right?

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

Yep. Done.

 
   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 {
+
+class TaskSchedulerWorkerPoolSingleThreadedTest
+    : public TaskSchedulerWorkerPoolImplTest {
+ public:
+  void InitializeThreadChecker() {
+    thread_checker_.reset(new ThreadCheckerImpl());
+  }
+
+  void CheckValidThread() {
+    EXPECT_TRUE(thread_checker_->CalledOnValidThread());
+  }
+
+ protected:
+  void SetUp() override {
+    InitializeWorkerPool(kSuggestedReclaimTime);
+  }
+
+  TaskSchedulerWorkerPoolSingleThreadedTest() = default;
+
+ private:
+  std::unique_ptr<ThreadCheckerImpl> thread_checker_;
+
+  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::InitializeThreadChecker,
+           Unretained(this)));
+  WaitableEvent task_waiter(WaitableEvent::ResetPolicy::AUTOMATIC,
+                            WaitableEvent::InitialState::NOT_SIGNALED);
+  single_thread_task_runner->PostTask(
+      FROM_HERE, Bind(&WaitableEvent::Signal, Unretained(&task_waiter)));
+  task_waiter.Wait();
+  worker_pool_->WaitForAllWorkersIdleForTesting();
+
+  // Give the worker pool a chance to reclaim its threads.
+  PlatformThread::Sleep(
+      kSuggestedReclaimTime + TimeDelta::FromMilliseconds(200));
+
+  worker_pool_->DisallowWorkerDetachmentForTesting();
+
+  single_thread_task_runner->PostTask(
+      FROM_HERE,
+      Bind(&TaskSchedulerWorkerPoolSingleThreadedTest::CheckValidThread,
+           Unretained(this)));
+  single_thread_task_runner->PostTask(
+      FROM_HERE, Bind(&WaitableEvent::Signal, Unretained(&task_waiter)));
+  task_waiter.Wait();
+}
+
+namespace {
+
+constexpr size_t kMagicTlsValue = 42;
+
+class TaskSchedulerWorkerPoolCheckTlsReuse
+    : public TaskSchedulerWorkerPoolImplTest {
+ public:
+  void SetTlsValueAndWait() {
+    slot_.Set(reinterpret_cast<void*>(kMagicTlsValue));
+    waiter_.Wait();
+  }
+
+  void CountZeroTlsValuesAndWait(WaitableEvent* count_waiter) {

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

s/ZeroTlsValue/NoMagicTlsValue/ (here and below for variable?)

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

We're counting zero TLS values here so the name should stick. The TLS is always
initialized to zero.

+    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(kSuggestedReclaimTime);
+  }
+
+  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,
+               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);

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

Should this use NoBarrier_Load()? On Windows this is equivalent, but in
atomicops_internals_portable.h it appears to do more than just a simple deref.

Note: NoBarrier is clearly correct because of the surrounding use of
WaitableEvent (not that it's very efficient given the heavier synchronization
constructs nearby but the barrier is definitely not necessary)

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

atomicops.h:19  It is incorrect to make direct assignments to/from an atomic
variable. You should use one of the Load or Store routines.

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

The atomicops unittest direct references their variables for testing
(https://cs.chromium.org/chromium/src/base/atomicops_unittest.cc).

Went ahead and changed this to a NoBarrier_Load

+
+  // Release tasks waiting on |waiter_|.
+  waiter_.Signal();
+}
+
 }  // namespace internal
 }  // namespace base