Add a timeout to Future

Future has a `Get(Duration)` method which allows us to wait synchronously until
a timeout.
Adding a timeout to the constructor allows the Future to time out
asynchronously.

PiperOrigin-RevId: 516982717

internal/platform/future.h

@@ -15,6 +15,8 @@
 #ifndef PLATFORM_PUBLIC_FUTURE_H_
 #define PLATFORM_PUBLIC_FUTURE_H_
 
+#include <utility>
+
 #include "internal/platform/settable_future.h"
 
 namespace nearby {
@@ -22,6 +24,13 @@ namespace nearby {
 template <typename T>
 class Future final {
  public:
+  // Default Future. Does not time out.
+  Future() : impl_(std::make_shared<SettableFuture<T>>()) {}
+
+  // Creates a Future with a timeout.
+  explicit Future(absl::Duration timeout)
+      : impl_(std::make_shared<SettableFuture<T>>(timeout)) {}
+
   virtual bool Set(T value) { return impl_->Set(std::move(value)); }
   virtual bool SetException(Exception exception) {
     return impl_->SetException(exception);
@@ -46,7 +55,7 @@ class Future final {
   // 2)
   // Future<bool> future = DoSomeAsyncWork(); // Returns future, but keeps copy.
   // if (future.Get().Ok()) { /*...*/ }
-  std::shared_ptr<SettableFuture<T>> impl_{new SettableFuture<T>()};
+  std::shared_ptr<SettableFuture<T>> impl_;
 };
 
 }  // namespace nearby

internal/platform/future_test.cc

@@ -17,6 +17,8 @@
 #include "gtest/gtest.h"
 #include "absl/time/clock.h"
 #include "absl/time/time.h"
+#include "internal/platform/count_down_latch.h"
+#include "internal/platform/direct_executor.h"
 #include "internal/platform/exception.h"
 #include "internal/platform/single_thread_executor.h"
 
@@ -232,4 +234,37 @@ TEST(FutureTest, AddListenerWhenAlreadySetExceptionCallsCallback) {
   EXPECT_EQ(call_count, 1);
 }
 
+TEST(FutureTest, TimeoutSetsException) {
+  Future<int> future(absl::Milliseconds(10));
+
+  EXPECT_EQ(future.Get().exception(), Exception::kTimeout);
+}
+
+TEST(FutureTest, TimeoutCallsListeners) {
+  Future<int> future(absl::Milliseconds(10));
+  CountDownLatch latch(1);
+  future.AddListener([&]() { latch.CountDown(); },
+                     &DirectExecutor::GetInstance());
+
+  EXPECT_TRUE(latch.Await().Ok());
+
+  EXPECT_EQ(future.Get().exception(), Exception::kTimeout);
+}
+
+TEST(FutureTest, SetValueBeforeTimeout) {
+  Future<int> future(absl::Minutes(1));
+
+  future.Set(5);
+
+  EXPECT_EQ(future.Get().result(), 5);
+}
+
+TEST(FutureTest, SetExceptionBeforeTimeout) {
+  Future<int> future(absl::Minutes(1));
+
+  future.SetException({Exception::kExecution});
+
+  EXPECT_EQ(future.Get().exception(), Exception::kExecution);
+}
+
 }  // namespace nearby

internal/platform/implementation/g3/BUILD

@@ -42,6 +42,7 @@ cc_library(
         "//internal/platform/implementation:types",
         "//internal/platform/implementation/shared:count_down_latch",
         "//internal/platform/implementation/shared:posix_mutex",
+        "@com_google_absl//absl/base:core_headers",
         "@com_google_absl//absl/container:flat_hash_map",
         "@com_google_absl//absl/strings",
         "@com_google_absl//absl/synchronization",

internal/platform/implementation/g3/timer.h

@@ -15,8 +15,13 @@
 #ifndef PLATFORM_IMPL_G3_TIMER_H_
 #define PLATFORM_IMPL_G3_TIMER_H_
 
+#include <atomic>
+#include <memory>
 #include <utility>
 
+#include "absl/base/thread_annotations.h"
+#include "absl/time/time.h"
+#include "internal/platform/implementation/g3/scheduled_executor.h"
 #include "internal/platform/implementation/timer.h"
 
 namespace nearby {
@@ -29,39 +34,55 @@ class Timer : public api::Timer {
 
   bool Create(int delay, int interval,
               absl::AnyInvocable<void()> callback) override {
-    if (delay < 0 || interval < 0) {
+    if (delay < 0 || interval < 0 || callback == nullptr) {
       return false;
     }
+    interval_ = absl::Milliseconds(interval);
     callback_ = std::move(callback);
     is_stopped_ = false;
-    return true;
+    return Schedule(absl::Milliseconds(delay));
   }
 
   bool Stop() override {
     is_stopped_ = true;
-    return true;
+    absl::MutexLock lock(&mutex_);
+    if (task_) {
+      bool result = task_->Cancel();
+      task_.reset();
+      return result;
+    }
+    return false;
   }
 
   bool FireNow() override {
-    if (is_stopped_ || !callback_) {
+    if (is_stopped_) {
       return false;
     }
     callback_();
     return true;
   }
 
-  // Mocked methods for test only
-  void TriggerCallback() {
-    if (is_stopped_ || callback_ == nullptr) {
-      return;
-    }
+ private:
+  bool Schedule(absl::Duration delay) {
+    if (delay == absl::ZeroDuration()) return false;
+    absl::MutexLock lock(&mutex_);
+    task_ = executor_.Schedule([this]() { TriggerCallback(); }, delay);
+    return true;
+  }
 
+  void TriggerCallback() {
+    if (is_stopped_) return;
+    Schedule(interval_);
     callback_();
   }
 
  private:
+  absl::Mutex mutex_;
   absl::AnyInvocable<void()> callback_;
-  bool is_stopped_ = false;
+  std::atomic_bool is_stopped_;
+  absl::Duration interval_;
+  std::shared_ptr<api::Cancelable> task_ ABSL_GUARDED_BY(mutex_);
+  ScheduledExecutor executor_;
 };
 
 }  // namespace g3

internal/platform/settable_future.h

@@ -15,6 +15,7 @@
 #ifndef PLATFORM_PUBLIC_SETTABLE_FUTURE_H_
 #define PLATFORM_PUBLIC_SETTABLE_FUTURE_H_
 
+#include <memory>
 #include <utility>
 #include <vector>
 
@@ -22,6 +23,7 @@
 #include "internal/platform/mutex.h"
 #include "internal/platform/mutex_lock.h"
 #include "internal/platform/system_clock.h"
+#include "internal/platform/timer_impl.h"
 
 namespace nearby {
 
@@ -29,10 +31,18 @@ template <typename T>
 class SettableFuture : public api::SettableFuture<T> {
  public:
   SettableFuture() = default;
+
+  // Creates a SettableFuture that fails with a kTimeout when `timeout` expires.
+  explicit SettableFuture(absl::Duration timeout)
+      : timer_(absl::make_unique<TimerImpl>()) {
+    timer_->Start(absl::ToInt64Milliseconds(timeout), 0,
+                  [this] { SetException({Exception::kTimeout}); });
+  }
   ~SettableFuture() override = default;
 
   bool Set(T value) override {
     MutexLock lock(&mutex_);
+    timer_.reset();
     if (!done_) {
       value_ = std::move(value);
       done_ = true;
@@ -60,6 +70,13 @@ class SettableFuture : public api::SettableFuture<T> {
 
   bool SetException(Exception exception) override {
     MutexLock lock(&mutex_);
+    if (timer_) {
+      timer_->Stop();
+      // We can't destroy the timer from the timer.
+      if (!exception.Raised(Exception::kTimeout)) {
+        timer_.reset();
+      }
+    }
     return SetExceptionLocked(exception);
   }
 
@@ -120,6 +137,7 @@ class SettableFuture : public api::SettableFuture<T> {
   bool done_{false};
   T value_;
   Exception exception_{Exception::kFailed};
+  std::unique_ptr<TimerImpl> timer_;
 };
 
 }  // namespace nearby