Adds simulated time controller

This CL introduces the TimeControllerInterface that provides timing
related functionality. Most notably it provides a TaskQueueFactory
and facilitates creation of ProcessThread.

Two implementations of the interface are provided, RealTimeController
and SimulatedTimeController.

This prepares for an upcoming CL using these in Scenario tests.

Bug: webrtc:10365
Change-Id: Id956a29628d7e2f53ecaedadd643a9f697329d2f
Reviewed-on: https://webrtc-review.googlesource.com/c/src/+/127297
Commit-Queue: Sebastian Jansson <srte@webrtc.org>
Reviewed-by: Niels Moller <nisse@webrtc.org>
Cr-Commit-Position: refs/heads/master@{#27244}

test/BUILD.gn

@@ -392,6 +392,7 @@ if (rtc_include_tests) {
       "../test:single_threaded_task_queue",
       "pc/e2e:e2e_unittests",
       "scenario:scenario_unittests",
+      "time_controller:time_controller_unittests",
       "//testing/gmock",
       "//testing/gtest",
       "//third_party/abseil-cpp/absl/memory",

test/time_controller/BUILD.gn

@@ -0,0 +1,54 @@
+# Copyright (c) 2019 The WebRTC project authors. All Rights Reserved.
+#
+# Use of this source code is governed by a BSD-style license
+# that can be found in the LICENSE file in the root of the source
+# tree. An additional intellectual property rights grant can be found
+# in the file PATENTS.  All contributing project authors may
+# be found in the AUTHORS file in the root of the source tree.
+
+import("../../webrtc.gni")
+
+if (rtc_include_tests) {
+  rtc_source_set("time_controller") {
+    testonly = true
+    sources = [
+      "real_time_controller.cc",
+      "real_time_controller.h",
+      "simulated_time_controller.cc",
+      "simulated_time_controller.h",
+      "time_controller.h",
+    ]
+
+    deps = [
+      "../../api/task_queue",
+      "../../api/task_queue:global_task_queue_factory",
+      "../../api/units:time_delta",
+      "../../api/units:timestamp",
+      "../../modules:module_api",
+      "../../modules/utility:utility",
+      "../../rtc_base",
+      "../../rtc_base:rtc_base_tests_utils",
+      "../../rtc_base:rtc_event",
+      "../../rtc_base:sequenced_task_checker",
+      "../../rtc_base/synchronization:yield_policy",
+      "../../rtc_base/task_utils:to_queued_task",
+      "../../system_wrappers",
+      "//third_party/abseil-cpp/absl/memory",
+      "//third_party/abseil-cpp/absl/strings",
+    ]
+  }
+  rtc_source_set("time_controller_unittests") {
+    testonly = true
+    sources = [
+      "simulated_time_controller_unittest.cc",
+    ]
+    deps = [
+      ":time_controller",
+      "../:test_support",
+      "../../rtc_base:rtc_base_approved",
+      "../../rtc_base:rtc_task_queue",
+      "../../rtc_base/task_utils:repeating_task",
+      "//third_party/abseil-cpp/absl/memory",
+    ]
+  }
+}

test/time_controller/real_time_controller.cc

@@ -0,0 +1,41 @@
+/*
+ *  Copyright 2019 The WebRTC project authors. All Rights Reserved.
+ *
+ *  Use of this source code is governed by a BSD-style license
+ *  that can be found in the LICENSE file in the root of the source
+ *  tree. An additional intellectual property rights grant can be found
+ *  in the file PATENTS.  All contributing project authors may
+ *  be found in the AUTHORS file in the root of the source tree.
+ */
+#include "test/time_controller/real_time_controller.h"
+
+#include "api/task_queue/global_task_queue_factory.h"
+#include "rtc_base/event.h"
+#include "rtc_base/task_utils/to_queued_task.h"
+#include "system_wrappers/include/sleep.h"
+
+namespace webrtc {
+
+Clock* RealTimeController::GetClock() {
+  return Clock::GetRealTimeClock();
+}
+
+TaskQueueFactory* RealTimeController::GetTaskQueueFactory() {
+  return &GlobalTaskQueueFactory();
+}
+
+std::unique_ptr<ProcessThread> RealTimeController::CreateProcessThread(
+    const char* thread_name) {
+  return ProcessThread::Create(thread_name);
+}
+
+void RealTimeController::Sleep(TimeDelta duration) {
+  SleepMs(duration.ms());
+}
+
+void RealTimeController::InvokeWithControlledYield(
+    std::function<void()> closure) {
+  closure();
+}
+
+}  // namespace webrtc

test/time_controller/real_time_controller.h

@@ -0,0 +1,30 @@
+/*
+ *  Copyright 2019 The WebRTC project authors. All Rights Reserved.
+ *
+ *  Use of this source code is governed by a BSD-style license
+ *  that can be found in the LICENSE file in the root of the source
+ *  tree. An additional intellectual property rights grant can be found
+ *  in the file PATENTS.  All contributing project authors may
+ *  be found in the AUTHORS file in the root of the source tree.
+ */
+#ifndef TEST_TIME_CONTROLLER_REAL_TIME_CONTROLLER_H_
+#define TEST_TIME_CONTROLLER_REAL_TIME_CONTROLLER_H_
+
+#include <memory>
+
+#include "test/time_controller/time_controller.h"
+
+namespace webrtc {
+class RealTimeController : public TimeController {
+ public:
+  Clock* GetClock() override;
+  TaskQueueFactory* GetTaskQueueFactory() override;
+  std::unique_ptr<ProcessThread> CreateProcessThread(
+      const char* thread_name) override;
+  void Sleep(TimeDelta duration) override;
+  void InvokeWithControlledYield(std::function<void()> closure) override;
+};
+
+}  // namespace webrtc
+
+#endif  // TEST_TIME_CONTROLLER_REAL_TIME_CONTROLLER_H_

test/time_controller/simulated_time_controller.cc

@@ -0,0 +1,415 @@
+/*
+ *  Copyright 2019 The WebRTC project authors. All Rights Reserved.
+ *
+ *  Use of this source code is governed by a BSD-style license
+ *  that can be found in the LICENSE file in the root of the source
+ *  tree. An additional intellectual property rights grant can be found
+ *  in the file PATENTS.  All contributing project authors may
+ *  be found in the AUTHORS file in the root of the source tree.
+ */
+#include "test/time_controller/simulated_time_controller.h"
+
+#include <algorithm>
+#include <deque>
+#include <list>
+#include <map>
+#include <set>
+#include <string>
+#include <thread>
+#include <vector>
+
+#include "absl/memory/memory.h"
+#include "absl/strings/string_view.h"
+
+namespace webrtc {
+
+namespace sim_time_impl {
+class SimulatedSequenceRunner : public ProcessThread, public TaskQueueBase {
+ public:
+  SimulatedSequenceRunner(SimulatedTimeControllerImpl* handler,
+                          absl::string_view queue_name)
+      : handler_(handler), name_(queue_name) {}
+  ~SimulatedSequenceRunner() override { handler_->Unregister(this); }
+
+  // Provides next run time.
+  Timestamp GetNextRunTime() const;
+
+  // Iterates through delayed tasks and modules and moves them to the ready set
+  // if they are supposed to execute by |at time|.
+  void UpdateReady(Timestamp at_time);
+  // Runs all ready tasks and modules and updates next run time.
+  void Run(Timestamp at_time);
+
+  // TaskQueueBase interface
+  void Delete() override;
+  // Note: PostTask is also in ProcessThread interface.
+  void PostTask(std::unique_ptr<QueuedTask> task) override;
+  void PostDelayedTask(std::unique_ptr<QueuedTask> task,
+                       uint32_t milliseconds) override;
+
+  // ProcessThread interface
+  void Start() override;
+  void Stop() override;
+  void WakeUp(Module* module) override;
+  void RegisterModule(Module* module, const rtc::Location& from) override;
+  void DeRegisterModule(Module* module) override;
+
+ private:
+  Timestamp GetCurrentTime() const { return handler_->CurrentTime(); }
+  void RunReadyTasks(Timestamp at_time) RTC_LOCKS_EXCLUDED(lock_);
+  void RunReadyModules(Timestamp at_time) RTC_EXCLUSIVE_LOCKS_REQUIRED(lock_);
+  void UpdateNextRunTime() RTC_EXCLUSIVE_LOCKS_REQUIRED(lock_);
+
+  SimulatedTimeControllerImpl* const handler_;
+  const std::string name_;
+
+  rtc::CriticalSection lock_;
+
+  std::deque<std::unique_ptr<QueuedTask>> ready_tasks_ RTC_GUARDED_BY(lock_);
+  std::multimap<Timestamp, std::unique_ptr<QueuedTask>> delayed_tasks_
+      RTC_GUARDED_BY(lock_);
+
+  bool process_thread_running_ RTC_GUARDED_BY(lock_) = false;
+  std::set<Module*> stopped_modules_ RTC_GUARDED_BY(lock_);
+  std::set<Module*> ready_modules_ RTC_GUARDED_BY(lock_);
+  std::multimap<Timestamp, Module*> delayed_modules_ RTC_GUARDED_BY(lock_);
+
+  Timestamp next_run_time_ RTC_GUARDED_BY(lock_) = Timestamp::PlusInfinity();
+};
+
+Timestamp SimulatedSequenceRunner::GetNextRunTime() const {
+  rtc::CritScope lock(&lock_);
+  return next_run_time_;
+}
+
+void SimulatedSequenceRunner::UpdateReady(Timestamp at_time) {
+  rtc::CritScope lock(&lock_);
+  for (auto it = delayed_tasks_.begin();
+       it != delayed_tasks_.end() && it->first <= at_time;) {
+    ready_tasks_.emplace_back(std::move(it->second));
+    it = delayed_tasks_.erase(it);
+  }
+  for (auto it = delayed_modules_.begin();
+       it != delayed_modules_.end() && it->first <= at_time;) {
+    ready_modules_.insert(it->second);
+    it = delayed_modules_.erase(it);
+  }
+}
+
+void SimulatedSequenceRunner::Run(Timestamp at_time) {
+  RunReadyTasks(at_time);
+  rtc::CritScope lock(&lock_);
+  RunReadyModules(at_time);
+  UpdateNextRunTime();
+}
+
+void SimulatedSequenceRunner::Delete() {
+  {
+    rtc::CritScope lock(&lock_);
+    ready_tasks_.clear();
+    delayed_tasks_.clear();
+  }
+  delete this;
+}
+
+void SimulatedSequenceRunner::RunReadyTasks(Timestamp at_time) {
+  std::deque<std::unique_ptr<QueuedTask>> ready_tasks;
+  {
+    rtc::CritScope lock(&lock_);
+    ready_tasks.swap(ready_tasks_);
+  }
+  if (!ready_tasks.empty()) {
+    CurrentTaskQueueSetter set_current(this);
+    for (auto& ready : ready_tasks) {
+      bool delete_task = ready->Run();
+      if (delete_task) {
+        ready.reset();
+      } else {
+        ready.release();
+      }
+    }
+  }
+}
+
+void SimulatedSequenceRunner::RunReadyModules(Timestamp at_time) {
+  if (!ready_modules_.empty()) {
+    CurrentTaskQueueSetter set_current(this);
+    for (auto* module : ready_modules_) {
+      module->Process();
+      Timestamp next_run_time =
+          at_time + TimeDelta::ms(module->TimeUntilNextProcess());
+      delayed_modules_.emplace(next_run_time, module);
+    }
+  }
+  ready_modules_.clear();
+}
+
+void SimulatedSequenceRunner::UpdateNextRunTime() {
+  if (!ready_tasks_.empty() || !ready_modules_.empty()) {
+    next_run_time_ = Timestamp::MinusInfinity();
+  } else {
+    next_run_time_ = Timestamp::PlusInfinity();
+    if (!delayed_tasks_.empty())
+      next_run_time_ = std::min(next_run_time_, delayed_tasks_.begin()->first);
+    if (!delayed_modules_.empty())
+      next_run_time_ =
+          std::min(next_run_time_, delayed_modules_.begin()->first);
+  }
+}
+
+void SimulatedSequenceRunner::PostTask(std::unique_ptr<QueuedTask> task) {
+  rtc::CritScope lock(&lock_);
+  ready_tasks_.emplace_back(std::move(task));
+  next_run_time_ = Timestamp::MinusInfinity();
+}
+
+void SimulatedSequenceRunner::PostDelayedTask(std::unique_ptr<QueuedTask> task,
+                                              uint32_t milliseconds) {
+  rtc::CritScope lock(&lock_);
+  Timestamp target_time = GetCurrentTime() + TimeDelta::ms(milliseconds);
+  delayed_tasks_.emplace(target_time, std::move(task));
+  next_run_time_ = std::min(next_run_time_, target_time);
+}
+
+void SimulatedSequenceRunner::Start() {
+  std::set<Module*> starting;
+  {
+    rtc::CritScope lock(&lock_);
+    if (process_thread_running_)
+      return;
+    process_thread_running_ = true;
+    starting.swap(stopped_modules_);
+  }
+  for (auto& module : starting)
+    module->ProcessThreadAttached(this);
+
+  Timestamp at_time = GetCurrentTime();
+  rtc::CritScope lock(&lock_);
+  for (auto& module : starting)
+    delayed_modules_.insert(
+        {at_time + TimeDelta::ms(module->TimeUntilNextProcess()), module});
+  UpdateNextRunTime();
+}
+
+void SimulatedSequenceRunner::Stop() {
+  std::set<Module*> stopping;
+  {
+    rtc::CritScope lock(&lock_);
+    process_thread_running_ = false;
+
+    for (auto* ready : ready_modules_)
+      stopped_modules_.insert(ready);
+    ready_modules_.clear();
+
+    for (auto& delayed : delayed_modules_)
+      stopped_modules_.insert(delayed.second);
+    delayed_modules_.clear();
+
+    stopping = stopped_modules_;
+  }
+  for (auto& module : stopping)
+    module->ProcessThreadAttached(nullptr);
+}
+
+void SimulatedSequenceRunner::WakeUp(Module* module) {
+  rtc::CritScope lock(&lock_);
+  // If we already are planning to run this module as soon as possible, we don't
+  // need to do anything.
+  if (ready_modules_.find(module) != ready_modules_.end())
+    return;
+
+  for (auto it = delayed_modules_.begin(); it != delayed_modules_.end(); ++it) {
+    if (it->second == module) {
+      delayed_modules_.erase(it);
+      break;
+    }
+  }
+  Timestamp next_time =
+      GetCurrentTime() + TimeDelta::ms(module->TimeUntilNextProcess());
+  delayed_modules_.insert({next_time, module});
+  next_run_time_ = std::min(next_run_time_, next_time);
+}
+
+void SimulatedSequenceRunner::RegisterModule(Module* module,
+                                             const rtc::Location& from) {
+  module->ProcessThreadAttached(this);
+  rtc::CritScope lock(&lock_);
+  if (!process_thread_running_) {
+    stopped_modules_.insert(module);
+  } else {
+    Timestamp next_time =
+        GetCurrentTime() + TimeDelta::ms(module->TimeUntilNextProcess());
+    delayed_modules_.insert({next_time, module});
+    next_run_time_ = std::min(next_run_time_, next_time);
+  }
+}
+
+void SimulatedSequenceRunner::DeRegisterModule(Module* module) {
+  bool modules_running;
+  {
+    rtc::CritScope lock(&lock_);
+    if (!process_thread_running_) {
+      stopped_modules_.erase(module);
+    } else {
+      ready_modules_.erase(module);
+      for (auto it = delayed_modules_.begin(); it != delayed_modules_.end();
+           ++it) {
+        if (it->second == module) {
+          delayed_modules_.erase(it);
+          break;
+        }
+      }
+    }
+    modules_running = process_thread_running_;
+  }
+  if (modules_running)
+    module->ProcessThreadAttached(nullptr);
+}
+
+SimulatedTimeControllerImpl::SimulatedTimeControllerImpl(Timestamp start_time)
+    : thread_id_(rtc::CurrentThreadId()), current_time_(start_time) {}
+
+SimulatedTimeControllerImpl::~SimulatedTimeControllerImpl() = default;
+
+std::unique_ptr<TaskQueueBase, TaskQueueDeleter>
+SimulatedTimeControllerImpl::CreateTaskQueue(
+    absl::string_view name,
+    TaskQueueFactory::Priority priority) const {
+  // TODO(srte): Remove the const cast when the interface is made mutable.
+  auto mutable_this = const_cast<SimulatedTimeControllerImpl*>(this);
+  auto task_queue = std::unique_ptr<SimulatedSequenceRunner, TaskQueueDeleter>(
+      new SimulatedSequenceRunner(mutable_this, name));
+  rtc::CritScope lock(&mutable_this->lock_);
+  mutable_this->runners_.insert(task_queue.get());
+  return task_queue;
+}
+
+std::unique_ptr<ProcessThread> SimulatedTimeControllerImpl::CreateProcessThread(
+    const char* thread_name) {
+  rtc::CritScope lock(&lock_);
+  auto process_thread =
+      absl::make_unique<SimulatedSequenceRunner>(this, thread_name);
+  runners_.insert(process_thread.get());
+  return process_thread;
+}
+
+std::vector<SimulatedSequenceRunner*>
+SimulatedTimeControllerImpl::GetNextReadyRunner(Timestamp current_time) {
+  rtc::CritScope lock(&lock_);
+  std::vector<SimulatedSequenceRunner*> ready;
+  for (auto* runner : runners_) {
+    if (yielded_.find(runner) == yielded_.end() &&
+        runner->GetNextRunTime() <= current_time) {
+      ready.push_back(runner);
+    }
+  }
+  return ready;
+}
+
+void SimulatedTimeControllerImpl::YieldExecution() {
+  if (rtc::CurrentThreadId() == thread_id_) {
+    RTC_DCHECK_RUN_ON(&thread_checker_);
+    // When we yield, we don't want to risk executing further tasks on the
+    // currently executing task queue. If there's a ready task that also yields,
+    // it's added to this set as well and only tasks on the remaining task
+    // queues are executed.
+    auto inserted = yielded_.insert(TaskQueueBase::Current());
+    RTC_DCHECK(inserted.second);
+    RunReadyRunners();
+    yielded_.erase(inserted.first);
+  }
+}
+
+void SimulatedTimeControllerImpl::RunReadyRunners() {
+  RTC_DCHECK_RUN_ON(&thread_checker_);
+  Timestamp current_time = CurrentTime();
+  // We repeat until we have no ready left to handle tasks posted by ready
+  // runners.
+  while (true) {
+    auto ready = GetNextReadyRunner(current_time);
+    if (ready.empty())
+      break;
+    for (auto* runner : ready) {
+      runner->UpdateReady(current_time);
+      runner->Run(current_time);
+    }
+  }
+}
+
+Timestamp SimulatedTimeControllerImpl::CurrentTime() const {
+  rtc::CritScope lock(&time_lock_);
+  return current_time_;
+}
+
+Timestamp SimulatedTimeControllerImpl::NextRunTime() const {
+  Timestamp current_time = CurrentTime();
+  Timestamp next_time = Timestamp::PlusInfinity();
+  rtc::CritScope lock(&lock_);
+  for (auto* runner : runners_) {
+    Timestamp next_run_time = runner->GetNextRunTime();
+    if (next_run_time <= current_time)
+      return current_time;
+    next_time = std::min(next_time, next_run_time);
+  }
+  return next_time;
+}
+
+void SimulatedTimeControllerImpl::AdvanceTime(Timestamp target_time) {
+  rtc::CritScope time_lock(&time_lock_);
+  RTC_DCHECK(target_time >= current_time_);
+  current_time_ = target_time;
+}
+
+void SimulatedTimeControllerImpl::Unregister(SimulatedSequenceRunner* runner) {
+  rtc::CritScope lock(&lock_);
+  RTC_CHECK(runners_.erase(runner));
+}
+
+}  // namespace sim_time_impl
+
+GlobalSimulatedTimeController::GlobalSimulatedTimeController(
+    Timestamp start_time)
+    : sim_clock_(start_time.us()), impl_(start_time) {
+  global_clock_.SetTimeMicros(start_time.us());
+}
+
+GlobalSimulatedTimeController::~GlobalSimulatedTimeController() = default;
+
+Clock* GlobalSimulatedTimeController::GetClock() {
+  return &sim_clock_;
+}
+
+TaskQueueFactory* GlobalSimulatedTimeController::GetTaskQueueFactory() {
+  return &impl_;
+}
+
+std::unique_ptr<ProcessThread>
+GlobalSimulatedTimeController::CreateProcessThread(const char* thread_name) {
+  return impl_.CreateProcessThread(thread_name);
+}
+
+void GlobalSimulatedTimeController::Sleep(TimeDelta duration) {
+  rtc::ScopedYieldPolicy yield_policy(&impl_);
+  Timestamp current_time = impl_.CurrentTime();
+  Timestamp target_time = current_time + duration;
+  RTC_DCHECK_EQ(current_time.us(), rtc::TimeMicros());
+  while (current_time < target_time) {
+    impl_.RunReadyRunners();
+    Timestamp next_time = std::min(impl_.NextRunTime(), target_time);
+    impl_.AdvanceTime(next_time);
+    auto delta = next_time - current_time;
+    current_time = next_time;
+    sim_clock_.AdvanceTimeMicroseconds(delta.us());
+    global_clock_.AdvanceTimeMicros(delta.us());
+  }
+}
+
+void GlobalSimulatedTimeController::InvokeWithControlledYield(
+    std::function<void()> closure) {
+  rtc::ScopedYieldPolicy yield_policy(&impl_);
+  closure();
+}
+
+// namespace sim_time_impl
+
+}  // namespace webrtc

test/time_controller/simulated_time_controller.h

@@ -0,0 +1,102 @@
+/*
+ *  Copyright 2019 The WebRTC project authors. All Rights Reserved.
+ *
+ *  Use of this source code is governed by a BSD-style license
+ *  that can be found in the LICENSE file in the root of the source
+ *  tree. An additional intellectual property rights grant can be found
+ *  in the file PATENTS.  All contributing project authors may
+ *  be found in the AUTHORS file in the root of the source tree.
+ */
+#ifndef TEST_TIME_CONTROLLER_SIMULATED_TIME_CONTROLLER_H_
+#define TEST_TIME_CONTROLLER_SIMULATED_TIME_CONTROLLER_H_
+
+#include <memory>
+#include <unordered_set>
+#include <utility>
+#include <vector>
+
+#include "api/units/timestamp.h"
+#include "modules/include/module.h"
+#include "modules/utility/include/process_thread.h"
+#include "rtc_base/critical_section.h"
+#include "rtc_base/fake_clock.h"
+#include "rtc_base/platform_thread_types.h"
+#include "rtc_base/synchronization/yield_policy.h"
+#include "rtc_base/thread_checker.h"
+#include "test/time_controller/time_controller.h"
+
+namespace webrtc {
+
+namespace sim_time_impl {
+class SimulatedSequenceRunner;
+
+class SimulatedTimeControllerImpl : public TaskQueueFactory,
+                                    public rtc::YieldInterface {
+ public:
+  explicit SimulatedTimeControllerImpl(Timestamp start_time);
+  ~SimulatedTimeControllerImpl() override;
+
+  std::unique_ptr<TaskQueueBase, TaskQueueDeleter> CreateTaskQueue(
+      absl::string_view name,
+      Priority priority) const override;
+
+  // Implements the YieldInterface by running ready tasks on all task queues,
+  // except that if this method is called from a task, the task queue running
+  // that task is skipped.
+  void YieldExecution() override;
+  // Create process thread with the name |thread_name|.
+  std::unique_ptr<ProcessThread> CreateProcessThread(const char* thread_name);
+  // Runs all runners in |runners_| that has tasks or modules ready for
+  // execution.
+  void RunReadyRunners();
+  // Return |current_time_|.
+  Timestamp CurrentTime() const;
+  // Return min of runner->GetNextRunTime() for runner in |runners_|.
+  Timestamp NextRunTime() const;
+  // Set |current_time_| to |target_time|.
+  void AdvanceTime(Timestamp target_time);
+  // Removes |runner| from |runners_|.
+  void Unregister(SimulatedSequenceRunner* runner);
+
+ private:
+  // Returns runners in |runners_| that are ready for execution.
+  std::vector<SimulatedSequenceRunner*> GetNextReadyRunner(
+      Timestamp current_time) RTC_RUN_ON(thread_checker_);
+
+  const rtc::PlatformThreadId thread_id_;
+  rtc::ThreadChecker thread_checker_;
+  rtc::CriticalSection time_lock_;
+  Timestamp current_time_ RTC_GUARDED_BY(time_lock_);
+  rtc::CriticalSection lock_;
+  std::unordered_set<SimulatedSequenceRunner*> runners_ RTC_GUARDED_BY(lock_);
+  // Task queues on which YieldExecution has been called.
+  std::unordered_set<TaskQueueBase*> yielded_ RTC_GUARDED_BY(thread_checker_);
+};
+}  // namespace sim_time_impl
+
+// TimeController implementation using completely simulated time. Task queues
+// and process threads created by this controller will run delayed activities
+// when Sleep() is called. Overrides the global clock backing rtc::TimeMillis()
+// and rtc::TimeMicros(). Note that this is not thread safe since it modifies
+// global state.
+class GlobalSimulatedTimeController : public TimeController {
+ public:
+  explicit GlobalSimulatedTimeController(Timestamp start_time);
+  ~GlobalSimulatedTimeController() override;
+
+  Clock* GetClock() override;
+  TaskQueueFactory* GetTaskQueueFactory() override;
+  std::unique_ptr<ProcessThread> CreateProcessThread(
+      const char* thread_name) override;
+  void Sleep(TimeDelta duration) override;
+  void InvokeWithControlledYield(std::function<void()> closure) override;
+
+ private:
+  rtc::ScopedFakeClock global_clock_;
+  // Provides simulated CurrentNtpInMilliseconds()
+  SimulatedClock sim_clock_;
+  sim_time_impl::SimulatedTimeControllerImpl impl_;
+};
+}  // namespace webrtc
+
+#endif  // TEST_TIME_CONTROLLER_SIMULATED_TIME_CONTROLLER_H_

test/time_controller/simulated_time_controller_unittest.cc

@@ -0,0 +1,120 @@
+/*
+ *  Copyright 2019 The WebRTC project authors. All Rights Reserved.
+ *
+ *  Use of this source code is governed by a BSD-style license
+ *  that can be found in the LICENSE file in the root of the source
+ *  tree. An additional intellectual property rights grant can be found
+ *  in the file PATENTS.  All contributing project authors may
+ *  be found in the AUTHORS file in the root of the source tree.
+ */
+
+#include <atomic>
+#include <memory>
+
+#include "absl/memory/memory.h"
+#include "rtc_base/task_queue.h"
+#include "rtc_base/task_utils/repeating_task.h"
+#include "test/gmock.h"
+#include "test/gtest.h"
+#include "test/time_controller/simulated_time_controller.h"
+
+// NOTE: Since these tests rely on real time behavior, they will be flaky
+// if run on heavily loaded systems.
+namespace webrtc {
+namespace {
+using ::testing::AtLeast;
+using ::testing::Invoke;
+using ::testing::MockFunction;
+using ::testing::NiceMock;
+using ::testing::Return;
+constexpr Timestamp kStartTime = Timestamp::Seconds<1000>();
+
+// Helper closure class to stop repeating task on a task queue. This is
+// equivalent to [handle{move(handle)}] { handle.Stop(); } in c++14.
+class TaskHandleStopper {
+ public:
+  explicit TaskHandleStopper(RepeatingTaskHandle handle)
+      : handle_(std::move(handle)) {}
+  void operator()() { handle_.Stop(); }
+
+ private:
+  RepeatingTaskHandle handle_;
+};
+}  // namespace
+
+TEST(SimulatedTimeControllerTest, TaskIsStoppedOnStop) {
+  const TimeDelta kShortInterval = TimeDelta::ms(5);
+  const TimeDelta kLongInterval = TimeDelta::ms(20);
+  const int kShortIntervalCount = 4;
+  const int kMargin = 1;
+  GlobalSimulatedTimeController time_simulation(kStartTime);
+  rtc::TaskQueue task_queue(
+      time_simulation.GetTaskQueueFactory()->CreateTaskQueue(
+          "TestQueue", TaskQueueFactory::Priority::NORMAL));
+  std::atomic_int counter(0);
+  auto handle = RepeatingTaskHandle::Start(task_queue.Get(), [&] {
+    if (++counter >= kShortIntervalCount)
+      return kLongInterval;
+    return kShortInterval;
+  });
+  // Sleep long enough to go through the initial phase.
+  time_simulation.Sleep(kShortInterval * (kShortIntervalCount + kMargin));
+  EXPECT_EQ(counter.load(), kShortIntervalCount);
+
+  task_queue.PostTask(TaskHandleStopper(std::move(handle)));
+  // Sleep long enough that the task would run at least once more if not
+  // stopped.
+  time_simulation.Sleep(kLongInterval * 2);
+  EXPECT_EQ(counter.load(), kShortIntervalCount);
+}
+
+TEST(SimulatedTimeControllerTest, TaskCanStopItself) {
+  std::atomic_int counter(0);
+  GlobalSimulatedTimeController time_simulation(kStartTime);
+  rtc::TaskQueue task_queue(
+      time_simulation.GetTaskQueueFactory()->CreateTaskQueue(
+          "TestQueue", TaskQueueFactory::Priority::NORMAL));
+
+  RepeatingTaskHandle handle;
+  task_queue.PostTask([&] {
+    handle = RepeatingTaskHandle::Start(task_queue.Get(), [&] {
+      ++counter;
+      handle.Stop();
+      return TimeDelta::ms(2);
+    });
+  });
+  time_simulation.Sleep(TimeDelta::ms(10));
+  EXPECT_EQ(counter.load(), 1);
+}
+TEST(SimulatedTimeControllerTest, Example) {
+  class ObjectOnTaskQueue {
+   public:
+    void DoPeriodicTask() {}
+    TimeDelta TimeUntilNextRun() { return TimeDelta::ms(100); }
+    void StartPeriodicTask(RepeatingTaskHandle* handle,
+                           rtc::TaskQueue* task_queue) {
+      *handle = RepeatingTaskHandle::Start(task_queue->Get(), [this] {
+        DoPeriodicTask();
+        return TimeUntilNextRun();
+      });
+    }
+  };
+  GlobalSimulatedTimeController time_simulation(kStartTime);
+  rtc::TaskQueue task_queue(
+      time_simulation.GetTaskQueueFactory()->CreateTaskQueue(
+          "TestQueue", TaskQueueFactory::Priority::NORMAL));
+  auto object = absl::make_unique<ObjectOnTaskQueue>();
+  // Create and start the periodic task.
+  RepeatingTaskHandle handle;
+  object->StartPeriodicTask(&handle, &task_queue);
+  // Restart the task
+  task_queue.PostTask(TaskHandleStopper(std::move(handle)));
+  object->StartPeriodicTask(&handle, &task_queue);
+  task_queue.PostTask(TaskHandleStopper(std::move(handle)));
+  struct Destructor {
+    void operator()() { object.reset(); }
+    std::unique_ptr<ObjectOnTaskQueue> object;
+  };
+  task_queue.PostTask(Destructor{std::move(object)});
+}
+}  // namespace webrtc

test/time_controller/time_controller.h

@@ -0,0 +1,47 @@
+/*
+ *  Copyright 2019 The WebRTC project authors. All Rights Reserved.
+ *
+ *  Use of this source code is governed by a BSD-style license
+ *  that can be found in the LICENSE file in the root of the source
+ *  tree. An additional intellectual property rights grant can be found
+ *  in the file PATENTS.  All contributing project authors may
+ *  be found in the AUTHORS file in the root of the source tree.
+ */
+#ifndef TEST_TIME_CONTROLLER_TIME_CONTROLLER_H_
+#define TEST_TIME_CONTROLLER_TIME_CONTROLLER_H_
+
+#include <functional>
+#include <memory>
+
+#include "api/task_queue/task_queue_factory.h"
+#include "api/units/time_delta.h"
+#include "modules/utility/include/process_thread.h"
+#include "system_wrappers/include/clock.h"
+
+namespace webrtc {
+
+// Interface for controlling time progress. This allows us to execute test code
+// in either real time or simulated time by using different implementation of
+// this interface.
+class TimeController {
+ public:
+  virtual ~TimeController() = default;
+  // Provides a clock instance that follows implementation defined time
+  // progress.
+  virtual Clock* GetClock() = 0;
+  // The returned factory will created task queues that runs in implementation
+  // defined time domain.
+  virtual TaskQueueFactory* GetTaskQueueFactory() = 0;
+  // Creates a process thread.
+  virtual std::unique_ptr<ProcessThread> CreateProcessThread(
+      const char* thread_name) = 0;
+  // Allow task queues and process threads created by this instance to execute
+  // for the given |duration|.
+  virtual void Sleep(TimeDelta duration) = 0;
+  // Execute closure in an implementation defined scope where rtc::Event::Wait
+  // might yield to execute other tasks. This allows doing blocking waits on
+  // tasks on other task queues froma a task queue without deadlocking.
+  virtual void InvokeWithControlledYield(std::function<void()> closure) = 0;
+};
+}  // namespace webrtc
+#endif  // TEST_TIME_CONTROLLER_TIME_CONTROLLER_H_