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,
+        kSuggestedReclaimTime,
         Bind(&TaskSchedulerWorkerPoolImplTest::ReEnqueueSequenceCallback,
              Unretained(this)),
         &task_tracker_, &delayed_task_manager_);
     ASSERT_TRUE(worker_pool_);
   }
 
   void TearDown() override {
     worker_pool_->WaitForAllWorkersIdleForTesting();
     worker_pool_->JoinForTesting();
   }
@@ skipping 279 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);
 
   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()));
+  }
+
+ protected:
+  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();
+  worker_pool_->WaitForAllWorkersIdleForTesting();
+
+  // Give the worker pool a chance to reclaim its threads.
+  PlatformThread::Sleep(
+      kSuggestedReclaimTime + TimeDelta::FromMilliseconds(200));
+
+  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();
+}
+

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

Test that no detachment occurs when suggested reclaim time is TimeDelta::Max?

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

If we could test this, this would be better handled in the SchedulerWorkerTest.
In either case, I'm not sure how we could test this since it depends on an
indefinite waiting period. The test would have to check the state at an
arbitrary time, and if the component was indeed adversarial, it could simply
detach after the arbitrary time. 

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

Acknowledged.

+namespace {
+
+class TaskSchedulerWorkerPoolCheckTlsReuse
+    : public TaskSchedulerWorkerPoolImplTest {
+ public:
+  void SetTlsValueAndWait() {
+    slot_.Set(reinterpret_cast<void*>(kMagicTlsValue));
+    waiter_.Wait();
+  }
+
+  void CountZeroTlsValues() {
+    if (!slot_.Get())
+      subtle::NoBarrier_AtomicIncrement(&zero_tls_values_, 1);
+  }
+
+ protected:
+  TaskSchedulerWorkerPoolCheckTlsReuse() :
+      waiter_(WaitableEvent::ResetPolicy::MANUAL,
+              WaitableEvent::InitialState::NOT_SIGNALED) {}
+  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();

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

Should the Reset() be after WaitForAllWorkersIdleForTesting()? A task might not
have reached line 457 yet when Reset() is callled.

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

The tricky bit here is that we can't wait for all idle workers because they
won't ever be idle if they're waiting for  the signal. The WaitForAllTasksToRun
in the for loop is the best current guarantee short of doing something
semaphore-ish to make sure the thread is occupied before dispatching the next
set.

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

If Signal() stays before WaitForAllWorkersIdleForTesting(), all workers should
eventually become idle? I think Reset() should be moved after
WaitForAllWorkersIdleForTesting() to avoid having a task that doesn't call
Wait() before Reset() is called (and thus blocks forever when it does call
Wait()).

[robliao, 2016/07/08 17:25:49]

I initially misread this. Yup, let's move Reset after the wait for idle.

+  waiter_.Reset();
+
+  worker_pool_->WaitForAllWorkersIdleForTesting();
+
+  // Give the worker pool a chance to detach its threads.
+  PlatformThread::Sleep(
+      kSuggestedReclaimTime + TimeDelta::FromMilliseconds(200));
+
+  // 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) {
+    ASSERT_TRUE(factory->PostTask(
+          PostNestedTask::NO,
+          Bind(&TaskSchedulerWorkerPoolCheckTlsReuse::CountZeroTlsValues,

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

Nothing prevents all CountZeroTlsValue tasks from running on the same thread
(the Wait task posted at line 522 doesn't have to run on the same thread as the
CountZeroTlsValue task). Maybe you should move the Signal+Wait inside
CountZeroTlsValue?

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

Doesn't WaitForAllTasksToRun with a waiting task also help provide this
guarantee?

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

Line 486: You create factories with PARALLEL TaskRunners.

Loop iteration 1:
Line 511: Task runs on thread A.
Lines 515 and 518: Tasks run on thread B (nothing prevents these tasks from
running on another thread than A).

Loop iteration 2:
Line 511: Task runs on thread A (again).
[since two CountZeroTlsValues tasks ran on the same thread, there is a thread
that won't be inspected]

[robliao, 2016/07/08 17:25:49]

Ah yes, indeed. I've now moved both waitable events into CountZeroTlsValues.

+               Unretained(this))));
+    count_waiters.push_back(WrapUnique(new WaitableEvent(
+        WaitableEvent::ResetPolicy::MANUAL,
+        WaitableEvent::InitialState::NOT_SIGNALED)));
+    ASSERT_TRUE(factory->PostTask(
+          PostNestedTask::NO,
+          Bind(&WaitableEvent::Signal,
+               Unretained(count_waiters.back().get()))));
+    ASSERT_TRUE(factory->PostTask(PostNestedTask::NO,
+                                  Bind(&WaitableEvent::Wait,
+                                       Unretained(&waiter_))));
+    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();
+  worker_pool_->WaitForAllWorkersIdleForTesting();
+}
+
 }  // namespace internal
 }  // namespace base