[Chromecast] Create ThreadHealthChecker class.

This can be used to make sure that tasks are being flushed
through a given task runner at a desired rate, a signal that can
be used to determine if tasks are blocking a thread.

BUG=Internal b/23908801
BUG=Internal b/36699101
TEST=cast_base_unittests

Change-Id: I22dc48134ac46b2bd00bccee8105a7580d1072c9
Reviewed-on: https://chromium-review.googlesource.com/664086
Commit-Queue: Simeon Anfinrud <sanfin@chromium.org>
Reviewed-by: Stephen Lanham <slan@chromium.org>
Cr-Commit-Position: refs/heads/master@{#501643}

chromecast/base/BUILD.gn

@@ -146,10 +146,12 @@ test("cast_base_unittests") {
     "process_utils_unittest.cc",
     "serializers_unittest.cc",
     "system_time_change_notifier_unittest.cc",
+    "thread_health_checker_unittest.cc",
   ]
 
   deps = [
     ":test_support",
+    ":thread_health_checker",
     "//base/test:run_all_unittests",
     "//base/test:test_support",
     "//testing/gmock",
@@ -211,6 +213,17 @@ source_set("cast_sys_info_shlib") {
   }
 }
 
+source_set("thread_health_checker") {
+  sources = [
+    "thread_health_checker.cc",
+    "thread_health_checker.h",
+  ]
+  deps = [
+    ":base",
+    "//base",
+  ]
+}
+
 process_version("cast_version") {
   template_file = "version.h.in"
   output = "$target_gen_dir/version.h"

chromecast/base/thread_health_checker.cc

+// Copyright 2017 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 "chromecast/base/thread_health_checker.h"
+
+#include <memory>
+#include <string>
+
+#include "base/memory/ptr_util.h"
+#include "base/memory/ref_counted.h"
+#include "base/single_thread_task_runner.h"
+#include "base/task_runner.h"
+#include "base/threading/thread_checker.h"
+#include "base/timer/timer.h"
+#include "chromecast/base/bind_to_task_runner.h"
+
+namespace chromecast {
+
+ThreadHealthChecker::Internal::Internal(
+    scoped_refptr<base::TaskRunner> patient_task_runner,
+    scoped_refptr<base::SingleThreadTaskRunner> doctor_task_runner,
+    base::TimeDelta interval,
+    base::TimeDelta timeout,
+    base::RepeatingClosure on_failure)
+    : patient_task_runner_(std::move(patient_task_runner)),
+      doctor_task_runner_(std::move(doctor_task_runner)),
+      interval_(interval),
+      timeout_(timeout),
+      on_failure_(std::move(on_failure)) {
+  DCHECK(patient_task_runner_);
+  DCHECK(doctor_task_runner_);
+}
+
+ThreadHealthChecker::Internal::~Internal() {}
+
+void ThreadHealthChecker::Internal::StartHealthCheck() {
+  DCHECK(doctor_task_runner_->BelongsToCurrentThread());
+  DETACH_FROM_THREAD(thread_checker_);
+  ok_timer_ = base::MakeUnique<base::OneShotTimer>();
+  failure_timer_ = base::MakeUnique<base::OneShotTimer>();
+  ScheduleHealthCheck();
+}
+
+void ThreadHealthChecker::Internal::StopHealthCheck() {
+  DCHECK_CALLED_ON_VALID_THREAD(thread_checker_);
+  DCHECK(ok_timer_);
+  DCHECK(failure_timer_);
+  ok_timer_->Stop();
+  failure_timer_->Stop();
+}
+
+void ThreadHealthChecker::Internal::ScheduleHealthCheck() {
+  DCHECK_CALLED_ON_VALID_THREAD(thread_checker_);
+  ok_timer_->Start(FROM_HERE, interval_, this,
+                   &ThreadHealthChecker::Internal::CheckThreadHealth);
+}
+
+void ThreadHealthChecker::Internal::CheckThreadHealth() {
+  DCHECK_CALLED_ON_VALID_THREAD(thread_checker_);
+  DCHECK(!failure_timer_->IsRunning());
+  patient_task_runner_->PostTask(
+      FROM_HERE, BindToCurrentThread(base::BindOnce(
+                     &ThreadHealthChecker::Internal::ThreadOk, this)));
+  failure_timer_->Start(FROM_HERE, timeout_, this,
+                        &ThreadHealthChecker::Internal::ThreadTimeout);
+}
+
+void ThreadHealthChecker::Internal::ThreadOk() {
+  DCHECK_CALLED_ON_VALID_THREAD(thread_checker_);
+  failure_timer_->Stop();
+  ScheduleHealthCheck();
+}
+
+void ThreadHealthChecker::Internal::ThreadTimeout() {
+  DCHECK_CALLED_ON_VALID_THREAD(thread_checker_);
+  on_failure_.Run();
+  ScheduleHealthCheck();
+}
+
+// The public ThreadHealthChecker owns a ref-counted reference to an Internal
+// object which does the heavy lifting.
+ThreadHealthChecker::ThreadHealthChecker(
+    scoped_refptr<base::TaskRunner> patient_task_runner,
+    scoped_refptr<base::SingleThreadTaskRunner> doctor_task_runner,
+    base::TimeDelta interval,
+    base::TimeDelta timeout,
+    base::RepeatingClosure on_failure)
+    : doctor_task_runner_(doctor_task_runner),
+      internal_(base::MakeRefCounted<ThreadHealthChecker::Internal>(
+          patient_task_runner,
+          doctor_task_runner,
+          interval,
+          timeout,
+          std::move(on_failure))) {
+  doctor_task_runner_->PostTask(
+      FROM_HERE,
+      base::BindOnce(&ThreadHealthChecker::Internal::StartHealthCheck,
+                     internal_));
+}
+
+// When the public ThreadHealthChecker is destroyed, the reference to the
+// Internal representation is freed, but if there are any pending tasks on the
+// doctor thread that partially own Internal, they will be run asynchronously
+// before the Internal object is destroyed and the health check stops.
+ThreadHealthChecker::~ThreadHealthChecker() {
+  doctor_task_runner_->PostTask(
+      FROM_HERE, base::BindOnce(&ThreadHealthChecker::Internal::StopHealthCheck,
+                                internal_));
+}
+
+}  // namespace chromecast

chromecast/base/thread_health_checker.h

+// Copyright 2017 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.
+
+#ifndef CHROMECAST_INTERNAL_SERVICE_THREAD_HEALTH_CHECKER_H_
+#define CHROMECAST_INTERNAL_SERVICE_THREAD_HEALTH_CHECKER_H_
+
+#include <memory>
+#include <string>
+
+#include "base/callback.h"
+#include "base/macros.h"
+#include "base/memory/ref_counted.h"
+#include "base/threading/thread_checker.h"
+#include "base/time/time.h"
+
+namespace base {
+class OneShotTimer;
+class SingleThreadTaskRunner;
+class TaskRunner;
+}  // namespace base
+
+namespace chromecast {
+// A class used to periodically check the responsiveness of a thread.
+//
+// The class takes two task runners, a "patient", and a "doctor". The doctor
+// task runner will post a "sentinel" task to the patient and verify that it
+// gets run within a certain response time, determined by |timeout|.
+// If the task is not run in time, then the patient fails the health checkup and
+// |on_failure| is invoked.
+//
+// The thread health is checked periodically, with the length between one check
+// and the next determined by |interval|, and the amount of time allowed for the
+// sentinel task to complete determined by |timeout|.
+class ThreadHealthChecker {
+ public:
+  ThreadHealthChecker(
+      scoped_refptr<base::TaskRunner> patient_task_runner,
+      scoped_refptr<base::SingleThreadTaskRunner> doctor_task_runner,
+      base::TimeDelta interval,
+      base::TimeDelta timeout,
+      base::RepeatingClosure on_failure);
+  ~ThreadHealthChecker();
+
+ private:
+  class Internal : public base::RefCountedThreadSafe<Internal> {
+   public:
+    Internal(scoped_refptr<base::TaskRunner> patient_task_runner,
+             scoped_refptr<base::SingleThreadTaskRunner> doctor_task_runner,
+             base::TimeDelta interval,
+             base::TimeDelta timeout,
+             base::RepeatingClosure on_failure);
+    void StartHealthCheck();
+    void StopHealthCheck();
+
+   private:
+    friend class base::RefCountedThreadSafe<Internal>;
+    ~Internal();
+    void ScheduleHealthCheck();
+    void CheckThreadHealth();
+    void ThreadOk();
+    void ThreadTimeout();
+
+    scoped_refptr<base::TaskRunner> patient_task_runner_;
+    scoped_refptr<base::SingleThreadTaskRunner> doctor_task_runner_;
+    base::TimeDelta interval_;
+    base::TimeDelta timeout_;
+    std::unique_ptr<base::OneShotTimer> ok_timer_;
+    std::unique_ptr<base::OneShotTimer> failure_timer_;
+    base::RepeatingClosure on_failure_;
+    THREAD_CHECKER(thread_checker_);
+  };
+
+  scoped_refptr<base::SingleThreadTaskRunner> doctor_task_runner_;
+  scoped_refptr<Internal> internal_;
+
+  DISALLOW_COPY_AND_ASSIGN(ThreadHealthChecker);
+};
+
+}  // namespace chromecast
+
+#endif  // CHROMECAST_INTERNAL_SERVICE_THREAD_HEALTH_CHECKER_H_

chromecast/base/thread_health_checker_unittest.cc

+// Copyright 2017 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 "chromecast/base/thread_health_checker.h"
+
+#include "base/memory/ref_counted.h"
+#include "base/synchronization/waitable_event.h"
+#include "base/test/test_mock_time_task_runner.h"
+#include "testing/gtest/include/gtest/gtest.h"
+
+namespace chromecast {
+
+namespace {
+const base::TimeDelta kInterval = base::TimeDelta::FromSeconds(3);
+const base::TimeDelta kTimeout = base::TimeDelta::FromSeconds(2);
+}  // namespace
+
+class ThreadHealthCheckerTest : public ::testing::Test {
+ protected:
+  ThreadHealthCheckerTest()
+      : patient_(base::MakeRefCounted<base::TestMockTimeTaskRunner>()),
+        doctor_(base::MakeRefCounted<base::TestMockTimeTaskRunner>()),
+        event_(base::WaitableEvent::ResetPolicy::MANUAL,
+               base::WaitableEvent::InitialState::NOT_SIGNALED) {}
+
+  ~ThreadHealthCheckerTest() override{};
+
+  scoped_refptr<base::TestMockTimeTaskRunner> patient_;
+  scoped_refptr<base::TestMockTimeTaskRunner> doctor_;
+  base::WaitableEvent event_;
+};
+
+#define CREATE_THREAD_HEALTH_CHECKER(name)                                   \
+  ThreadHealthChecker name(patient_, doctor_, kInterval, kTimeout,           \
+                           base::BindRepeating(&base::WaitableEvent::Signal, \
+                                               base::Unretained(&event_)))
+
+TEST_F(ThreadHealthCheckerTest, FiresTimeoutWhenTaskRunnerDoesNotFlush) {
+  CREATE_THREAD_HEALTH_CHECKER(thc);
+  // Do not flush the patient, so that the health check sentinel task won't run.
+  doctor_->FastForwardBy(base::TimeDelta::FromSeconds(6));
+  EXPECT_TRUE(event_.IsSignaled());
+}
+
+TEST_F(ThreadHealthCheckerTest, DoesNotFireTimeoutWhenTaskRunnerFlushesInTime) {
+  CREATE_THREAD_HEALTH_CHECKER(thc);
+  // Advance the doctor by enough time to post the health check, but not to time
+  // out.
+  doctor_->FastForwardBy(base::TimeDelta::FromSeconds(4));
+  // Advance the patient to let the sentinel task run.
+  patient_->FastForwardBy(base::TimeDelta::FromSeconds(4));
+  // Advance the doctor by enough time such that the sentinel not running would
+  // cause the failure callback to run.
+  doctor_->FastForwardBy(base::TimeDelta::FromSeconds(2));
+  EXPECT_FALSE(event_.IsSignaled());
+}
+
+TEST_F(ThreadHealthCheckerTest, FiresTimeoutWhenTaskRunnerFlushesTooLate) {
+  CREATE_THREAD_HEALTH_CHECKER(thc);
+  // Advance the doctor before the patient, to simulate a task in the patient
+  // that takes too long.
+  doctor_->FastForwardBy(base::TimeDelta::FromSeconds(6));
+  patient_->FastForwardBy(base::TimeDelta::FromSeconds(6));
+  // Flush the task runner so the health check sentinel task is run.
+  EXPECT_TRUE(event_.IsSignaled());
+}
+
+TEST_F(ThreadHealthCheckerTest, FiresTimeoutOnLaterIteration) {
+  CREATE_THREAD_HEALTH_CHECKER(thc);
+  // Advance the doctor enough to start the check.
+  doctor_->FastForwardBy(base::TimeDelta::FromSeconds(4));
+  // Advance the patient enough to run the task.
+  patient_->FastForwardBy(base::TimeDelta::FromSeconds(4));
+  // Advance the doctor enough to start the check again.
+  doctor_->FastForwardBy(base::TimeDelta::FromSeconds(4));
+  EXPECT_FALSE(event_.IsSignaled());
+  // Advance the doctor enough for the timeout from the second check to fire.
+  doctor_->FastForwardBy(base::TimeDelta::FromSeconds(2));
+  EXPECT_TRUE(event_.IsSignaled());
+}
+
+TEST_F(ThreadHealthCheckerTest, NoCrashWhenDestroyed) {
+  {
+    CREATE_THREAD_HEALTH_CHECKER(thc);
+    doctor_->RunUntilIdle();
+  }
+  doctor_->RunUntilIdle();
+}
+
+TEST_F(ThreadHealthCheckerTest, DropPendingEventsAfterDestruction) {
+  {
+    CREATE_THREAD_HEALTH_CHECKER(thc);
+    // Fast forward by enough time to have scheduled a health check.
+    doctor_->FastForwardBy(base::TimeDelta::FromSeconds(4));
+    EXPECT_FALSE(event_.IsSignaled());
+  }
+  // Fast forward by enough time for the health check to have executed.
+  // However, we want all pending events to be dropped after the destructor is
+  // called, so the event should not be signalled.
+  doctor_->FastForwardBy(base::TimeDelta::FromSeconds(2));
+  EXPECT_FALSE(event_.IsSignaled());
+}
+
+}  // namespace chromecast