diff --git a/internal/platform/future.h b/internal/platform/future.h index 07218b22..02e2cd1b 100644 --- a/internal/platform/future.h +++ b/internal/platform/future.h @@ -15,6 +15,8 @@ #ifndef PLATFORM_PUBLIC_FUTURE_H_ #define PLATFORM_PUBLIC_FUTURE_H_ +#include + #include "internal/platform/settable_future.h" namespace nearby { @@ -22,6 +24,13 @@ namespace nearby { template class Future final { public: + // Default Future. Does not time out. + Future() : impl_(std::make_shared>()) {} + + // Creates a Future with a timeout. + explicit Future(absl::Duration timeout) + : impl_(std::make_shared>(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 future = DoSomeAsyncWork(); // Returns future, but keeps copy. // if (future.Get().Ok()) { /*...*/ } - std::shared_ptr> impl_{new SettableFuture()}; + std::shared_ptr> impl_; }; } // namespace nearby diff --git a/internal/platform/future_test.cc b/internal/platform/future_test.cc index 7e6f151c..aaacd3e4 100644 --- a/internal/platform/future_test.cc +++ b/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 future(absl::Milliseconds(10)); + + EXPECT_EQ(future.Get().exception(), Exception::kTimeout); +} + +TEST(FutureTest, TimeoutCallsListeners) { + Future 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 future(absl::Minutes(1)); + + future.Set(5); + + EXPECT_EQ(future.Get().result(), 5); +} + +TEST(FutureTest, SetExceptionBeforeTimeout) { + Future future(absl::Minutes(1)); + + future.SetException({Exception::kExecution}); + + EXPECT_EQ(future.Get().exception(), Exception::kExecution); +} + } // namespace nearby diff --git a/internal/platform/implementation/g3/BUILD b/internal/platform/implementation/g3/BUILD index a7e0ac76..ad8ecd93 100644 --- a/internal/platform/implementation/g3/BUILD +++ b/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", diff --git a/internal/platform/implementation/g3/timer.h b/internal/platform/implementation/g3/timer.h index c72dfbac..0bd0bb89 100644 --- a/internal/platform/implementation/g3/timer.h +++ b/internal/platform/implementation/g3/timer.h @@ -15,8 +15,13 @@ #ifndef PLATFORM_IMPL_G3_TIMER_H_ #define PLATFORM_IMPL_G3_TIMER_H_ +#include +#include #include +#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 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 callback_; - bool is_stopped_ = false; + std::atomic_bool is_stopped_; + absl::Duration interval_; + std::shared_ptr task_ ABSL_GUARDED_BY(mutex_); + ScheduledExecutor executor_; }; } // namespace g3 diff --git a/internal/platform/settable_future.h b/internal/platform/settable_future.h index 89e0659b..12dbf763 100644 --- a/internal/platform/settable_future.h +++ b/internal/platform/settable_future.h @@ -15,6 +15,7 @@ #ifndef PLATFORM_PUBLIC_SETTABLE_FUTURE_H_ #define PLATFORM_PUBLIC_SETTABLE_FUTURE_H_ +#include #include #include @@ -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 class SettableFuture : public api::SettableFuture { public: SettableFuture() = default; + + // Creates a SettableFuture that fails with a kTimeout when `timeout` expires. + explicit SettableFuture(absl::Duration timeout) + : timer_(absl::make_unique()) { + 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 { 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 { bool done_{false}; T value_; Exception exception_{Exception::kFailed}; + std::unique_ptr timer_; }; } // namespace nearby