Support fake clock in test environment

A synthetic clock helps testing (long) timeouts by allowing us to move the time
forward by an arbitrary amount.

PiperOrigin-RevId: 520407442

internal/platform/BUILD

@@ -218,6 +218,7 @@ cc_library(
         ":types",
         ":uuid",
         "//internal/platform/implementation:comm",
+        "//internal/test",
         "@com_google_absl//absl/container:flat_hash_map",
         "@com_google_absl//absl/container:flat_hash_set",
         "@com_google_absl//absl/status",

internal/platform/implementation/g3/BUILD

@@ -37,11 +37,15 @@ cc_library(
     visibility = ["//visibility:private"],
     deps = [
         "//internal/platform:base",
+        "//internal/platform:logging",
+        "//internal/platform:test_util",
         "//internal/platform:util",
         "//internal/platform/implementation:types",
         "//internal/platform/implementation/shared:count_down_latch",
         "//internal/platform/implementation/shared:posix_mutex",
+        "//internal/test",
         "@com_google_absl//absl/base:core_headers",
+        "@com_google_absl//absl/container:btree",
         "@com_google_absl//absl/container:flat_hash_map",
         "@com_google_absl//absl/strings",
         "@com_google_absl//absl/synchronization",

internal/platform/implementation/g3/scheduled_executor.cc

@@ -18,9 +18,10 @@
 #include <memory>
 #include <utility>
 
-#include "absl/time/clock.h"
 #include "internal/platform/implementation/cancelable.h"
+#include "internal/platform/medium_environment.h"
 #include "internal/platform/runnable.h"
+#include "internal/test/fake_clock.h"
 
 namespace nearby {
 namespace g3 {
@@ -59,20 +60,71 @@ class ScheduledCancelable : public api::Cancelable {
 
 }  // namespace
 
+ScheduledExecutor::ScheduledExecutor() {
+  absl::optional<FakeClock*> fake_clock =
+      MediumEnvironment::Instance().GetSimulatedClock();
+  if (fake_clock.has_value()) {
+    name_ = absl::StrFormat("G3 scheduled executor %p", this);
+    (*fake_clock)->AddObserver(name_, [this]() { RunReadyTasks(); });
+  }
+}
+
+ScheduledExecutor::~ScheduledExecutor() {
+  absl::optional<FakeClock*> fake_clock =
+      MediumEnvironment::Instance().GetSimulatedClock();
+  if (fake_clock.has_value()) {
+    (*fake_clock)->RemoveObserver(name_);
+  }
+  executor_.Shutdown();
+}
+
 std::shared_ptr<api::Cancelable> ScheduledExecutor::Schedule(
     Runnable&& runnable, absl::Duration delay) {
   auto scheduled_cancelable = std::make_shared<ScheduledCancelable>();
   if (executor_.InShutdown()) {
     return scheduled_cancelable;
   }
-  executor_.ScheduleAfter(delay, [this, scheduled_cancelable,
-                                  runnable(std::move(runnable))]() mutable {
+  Runnable task = [this, scheduled_cancelable,
+                   runnable = std::move(runnable)]() mutable {
     if (!executor_.InShutdown() && scheduled_cancelable->MarkExecuted()) {
       runnable();
     }
-  });
+  };
+  absl::optional<FakeClock*> fake_clock =
+      MediumEnvironment::Instance().GetSimulatedClock();
+  if (fake_clock.has_value()) {
+    absl::Time trigger_time = (*fake_clock)->Now() + delay;
+    absl::MutexLock lock(&mutex_);
+    tasks_.insert(std::pair<absl::Time, std::unique_ptr<Runnable>>(
+        trigger_time, std::make_unique<Runnable>(std::move(task))));
+  } else {
+    executor_.ScheduleAfter(delay, std::move(task));
+  }
   return scheduled_cancelable;
 }
 
+void ScheduledExecutor::RunReadyTasks() {
+  absl::optional<FakeClock*> fake_clock =
+      MediumEnvironment::Instance().GetSimulatedClock();
+  if (executor_.InShutdown()) {
+    return;
+  }
+  if (!fake_clock.has_value()) {
+    return;
+  }
+  absl::Time current_time = (*fake_clock)->Now();
+  absl::MutexLock lock(&mutex_);
+  for (auto it = tasks_.begin(); it != tasks_.end();) {
+    if (it->first <= current_time) {
+      executor_.Execute(
+          [task = std::move(it->second)]() mutable { (*task)(); });
+      it = tasks_.erase(it);
+    } else {
+      // Tasks are sorted. We can stop iterating.
+      break;
+    }
+  }
+}
+
 }  // namespace g3
 }  // namespace nearby

internal/platform/implementation/g3/scheduled_executor.h

@@ -17,13 +17,15 @@
 
 #include <atomic>
 #include <memory>
+#include <string>
+#include <utility>
 
-#include "absl/time/clock.h"
+#include "absl/container/btree_map.h"
+#include "absl/time/time.h"
 #include "internal/platform/implementation/cancelable.h"
+#include "internal/platform/implementation/g3/single_thread_executor.h"
 #include "internal/platform/implementation/scheduled_executor.h"
 #include "internal/platform/runnable.h"
-#include "internal/platform/implementation/g3/single_thread_executor.h"
-#include "nisaba/port/thread_pool.h"
 
 namespace nearby {
 namespace g3 {
@@ -32,8 +34,8 @@ namespace g3 {
 // unbounded queue.
 class ScheduledExecutor final : public api::ScheduledExecutor {
  public:
-  ScheduledExecutor() = default;
-  ~ScheduledExecutor() override { executor_.Shutdown(); }
+  ScheduledExecutor();
+  ~ScheduledExecutor() override;
 
   void Execute(Runnable&& runnable) override {
     executor_.Execute(std::move(runnable));
@@ -43,7 +45,12 @@ class ScheduledExecutor final : public api::ScheduledExecutor {
   void Shutdown() override { executor_.Shutdown(); }
 
  private:
+  void RunReadyTasks();
   SingleThreadExecutor executor_;
+  std::string name_;
+  absl::Mutex mutex_;
+  absl::btree_multimap<absl::Time, std::unique_ptr<Runnable>> tasks_
+      ABSL_GUARDED_BY(mutex_);
 };
 
 }  // namespace g3

internal/platform/implementation/g3/system_clock.cc

@@ -16,12 +16,28 @@
 
 #include "absl/time/clock.h"
 #include "internal/platform/exception.h"
+#include "internal/platform/medium_environment.h"
+#include "internal/test/fake_clock.h"
 
 namespace nearby {
 
-absl::Time SystemClock::ElapsedRealtime() { return absl::Now(); }
+absl::Time SystemClock::ElapsedRealtime() {
+  absl::optional<FakeClock*> fake_clock =
+      MediumEnvironment::Instance().GetSimulatedClock();
+  if (fake_clock.has_value()) {
+    return (*fake_clock)->Now();
+  }
+  return absl::Now();
+}
+
 Exception SystemClock::Sleep(absl::Duration duration) {
-  absl::SleepFor(duration);
+  absl::optional<FakeClock*> fake_clock =
+      MediumEnvironment::Instance().GetSimulatedClock();
+  if (fake_clock.has_value()) {
+    (*fake_clock)->FastForward(duration);
+  } else {
+    absl::SleepFor(duration);
+  }
   return {Exception::kSuccess};
 }
 

internal/platform/medium_environment.cc

@@ -18,6 +18,7 @@
 #include <atomic>
 #include <cinttypes>
 #include <cstdint>
+#include <memory>
 #include <new>
 #include <string>
 #include <type_traits>
@@ -27,6 +28,7 @@
 #include "absl/container/flat_hash_set.h"
 #include "absl/status/status.h"
 #include "absl/strings/escaping.h"
+#include "absl/types/optional.h"
 #include "internal/platform/count_down_latch.h"
 #include "internal/platform/feature_flags.h"
 #include "internal/platform/implementation/ble_v2.h"
@@ -47,6 +49,9 @@ void MediumEnvironment::Start(EnvironmentConfig config) {
   if (!enabled_.exchange(true)) {
     NEARBY_LOGS(INFO) << "MediumEnvironment::Start()";
     config_ = std::move(config);
+    if (config_.use_simulated_clock) {
+      simulated_clock_ = std::make_unique<FakeClock>();
+    }
     Reset();
   }
 }
@@ -55,6 +60,8 @@ void MediumEnvironment::Stop() {
   if (enabled_.exchange(false)) {
     NEARBY_LOGS(INFO) << "MediumEnvironment::Stop()";
     Sync(false);
+    config_ = {};
+    simulated_clock_.reset();
   }
 }
 
@@ -1197,4 +1204,11 @@ void MediumEnvironment::SetFeatureFlags(const FeatureFlags::Flags& flags) {
   const_cast<FeatureFlags&>(FeatureFlags::GetInstance()).SetFlags(flags);
 }
 
+absl::optional<FakeClock*> MediumEnvironment::GetSimulatedClock() {
+  if (simulated_clock_) {
+    return absl::optional<FakeClock*>(simulated_clock_.get());
+  }
+  return absl::nullopt;
+}
+
 }  // namespace nearby

internal/platform/medium_environment.h

@@ -30,6 +30,7 @@
 #include "internal/platform/implementation/bluetooth_adapter.h"
 #include "internal/platform/implementation/bluetooth_classic.h"
 #include "internal/platform/uuid.h"
+#include "internal/test/fake_clock.h"
 #ifndef NO_WEBRTC
 #include "internal/platform/implementation/webrtc.h"
 #endif
@@ -52,6 +53,11 @@ struct EnvironmentConfig {
   // This is currently set to false, due to http://b/139734036 that would lead
   // to flaky tests.
   bool webrtc_enabled = false;
+
+  // Installs a simulated clock, which can be used to test timeouts.
+  // The simulated clock is automatically picked up by SystemClock, Timer and
+  // ScheduledExecutor implementations.
+  bool use_simulated_clock = false;
 };
 
 // MediumEnvironment is a simulated environment which allows multiple instances
@@ -376,6 +382,8 @@ class MediumEnvironment {
 
   void SetFeatureFlags(const FeatureFlags::Flags& flags);
 
+  absl::optional<FakeClock*> GetSimulatedClock();
+
  private:
   struct BluetoothMediumContext {
     BluetoothDiscoveryCallback callback;
@@ -503,6 +511,7 @@ class MediumEnvironment {
 
   bool use_valid_peer_connection_ = true;
   absl::Duration peer_connection_latency_ = absl::ZeroDuration();
+  std::unique_ptr<FakeClock> simulated_clock_;
 };
 
 }  // namespace nearby

internal/platform/scheduled_executor_test.cc

@@ -22,6 +22,7 @@
 #include "absl/time/time.h"
 #include "internal/platform/count_down_latch.h"
 #include "internal/platform/exception.h"
+#include "internal/platform/medium_environment.h"
 
 namespace nearby {
 
@@ -206,6 +207,62 @@ TEST(ScheduledExecutorTest, ExecuteDuringShutdownFails) {
   executor.Shutdown();
 }
 
+TEST(ScheduledExecutorTest, SimulatedClockCanSchedule) {
+  MediumEnvironment::Instance().Start({.use_simulated_clock = true});
+  FakeClock* fake_clock =
+      MediumEnvironment::Instance().GetSimulatedClock().value();
+  ScheduledExecutor executor;
+  std::atomic_int value = 0;
+  CountDownLatch first_task_latch(1);
+  CountDownLatch second_task_latch(1);
+  // schedule job due in kLongDelay.
+  executor.Schedule(
+      [&]() {
+        EXPECT_EQ(value, 1);
+        value = 5;
+        first_task_latch.CountDown();
+      },
+      kLongDelay);
+  // schedule job due in kShortDelay; must fire before the first one.
+  executor.Schedule(
+      [&]() {
+        EXPECT_EQ(value, 0);
+        value = 1;
+        second_task_latch.CountDown();
+      },
+      kShortDelay);
+  EXPECT_EQ(value, 0);
+  fake_clock->FastForward(kShortDelay - absl::Milliseconds(1));
+  EXPECT_EQ(value, 0);
+  fake_clock->FastForward(absl::Milliseconds(1));
+  second_task_latch.Await();
+  EXPECT_EQ(value, 1);
+  fake_clock->FastForward(kLongDelay - kShortDelay);
+  first_task_latch.Await();
+  EXPECT_EQ(value, 5);
+  // Very long sleep to make sure that the sleep is truly simulated.
+  fake_clock->FastForward(absl::Minutes(30));
+  MediumEnvironment::Instance().Stop();
+}
+
+TEST(ScheduledExecutorTest,
+     DestroyExecutorWithSimulatedClockIgnoresPendingTasks) {
+  MediumEnvironment::Instance().Start({.use_simulated_clock = true});
+  FakeClock* fake_clock =
+      MediumEnvironment::Instance().GetSimulatedClock().value();
+  {
+    ScheduledExecutor executor;
+    executor.Schedule(
+        [&]() {
+          // This task should never be executed.
+          EXPECT_TRUE(false);
+        },
+        kShortDelay);
+  }
+  fake_clock->FastForward(absl::Minutes(30));
+  MediumEnvironment::Instance().Stop();
+}
+
 struct ThreadCheckTestClass {
   ScheduledExecutor executor;
   int value ABSL_GUARDED_BY(executor) = 0;