From 9eca355226730ba01eabddf21ab86fce0e7d19e2 Mon Sep 17 00:00:00 2001 From: Guogang Li Date: Thu, 9 Oct 2025 09:15:40 -0700 Subject: [PATCH] Internal cleanup PiperOrigin-RevId: 817215477 --- internal/platform/BUILD | 1 - internal/platform/future.h | 99 ++++++++--- internal/platform/future_test.cc | 156 +--------------- internal/platform/implementation/BUILD | 3 - internal/platform/implementation/future.h | 44 ----- .../implementation/listenable_future.h | 42 ----- internal/platform/implementation/platform.h | 2 - .../platform/implementation/settable_future.h | 47 ----- .../implementation/submittable_executor.h | 3 - internal/platform/settable_future.h | 168 ------------------ 10 files changed, 73 insertions(+), 492 deletions(-) delete mode 100644 internal/platform/implementation/future.h delete mode 100644 internal/platform/implementation/listenable_future.h delete mode 100644 internal/platform/implementation/settable_future.h delete mode 100644 internal/platform/settable_future.h diff --git a/internal/platform/BUILD b/internal/platform/BUILD index 8c5b9147..ef0c621f 100644 --- a/internal/platform/BUILD +++ b/internal/platform/BUILD @@ -241,7 +241,6 @@ cc_library( "pending_job_registry.h", "pipe.h", "scheduled_executor.h", - "settable_future.h", "single_thread_executor.h", "submittable_executor.h", "system_clock.h", diff --git a/internal/platform/future.h b/internal/platform/future.h index 9cd2a2c0..277eacb9 100644 --- a/internal/platform/future.h +++ b/internal/platform/future.h @@ -18,49 +18,92 @@ #include #include +#include "absl/base/thread_annotations.h" #include "absl/time/time.h" +#include "internal/platform/condition_variable.h" #include "internal/platform/exception.h" -#include "internal/platform/implementation/executor.h" -#include "internal/platform/settable_future.h" +#include "internal/platform/implementation/system_clock.h" +#include "internal/platform/mutex.h" +#include "internal/platform/mutex_lock.h" namespace nearby { template class Future final { public: - using FutureCallback = typename SettableFuture::FutureCallback; - // Default Future. Does not time out. - Future() : impl_(std::make_shared>()) {} + // Sets the value of the Future. + bool Set(T value) { + MutexLock lock(&state_->mutex); + if (!state_->done) { + state_->value = ExceptionOr(std::move(value)); + state_->done = true; + state_->completed.Notify(); + return true; + } + return false; + } - // Creates a Future with a timeout. - explicit Future(absl::Duration timeout) - : impl_(std::make_shared>(timeout)) {} + // Sets the exception of the Future. + bool SetException(Exception exception) { + MutexLock lock(&state_->mutex); + return SetExceptionLocked(exception); + } - virtual bool Set(T value) { return impl_->Set(std::move(value)); } - virtual bool SetException(Exception exception) { - return impl_->SetException(exception); + ExceptionOr Get() { + MutexLock lock(&state_->mutex); + while (!state_->done) { + state_->completed.Wait(); + } + return state_->value; } - virtual ExceptionOr Get() { return impl_->Get(); } - virtual ExceptionOr Get(absl::Duration timeout) { - return impl_->Get(timeout); + + // Gets the value of the Future, timing out after the specified duration. + ExceptionOr Get(absl::Duration timeout) { + MutexLock lock(&state_->mutex); + while (!state_->done) { + absl::Time start_time = SystemClock::ElapsedRealtime(); + if (state_->completed.Wait(timeout).Raised(Exception::kInterrupted)) { + SetExceptionLocked({Exception::kInterrupted}); + break; + } + absl::Duration spent = SystemClock::ElapsedRealtime() - start_time; + if (spent < timeout) { + timeout -= spent; + } else if (!state_->done) { + SetExceptionLocked({Exception::kTimeout}); + break; + } + } + return state_->value; } - void AddListener(FutureCallback callback, api::Executor* executor) { - impl_->AddListener(std::move(callback), executor); + + // Returns true if the Future has been set. + bool IsSet() const { + MutexLock lock(&state_->mutex); + return state_->done; } - bool IsSet() const { return impl_->IsSet(); } private: - // Instance of future implementation is wrapped in shared_ptr<> to make - // it possible to pass Future by value and share the implementation. - // This allows for the following constructions: - // 1) - // Future future; - // RunOnXyzThread([future]() { future.Set(DoTheJobAndReport()); }); - // if (future.Get().Ok()) { /*...*/ } - // 2) - // Future future = DoSomeAsyncWork(); // Returns future, but keeps copy. - // if (future.Get().Ok()) { /*...*/ } - std::shared_ptr> impl_; + struct FutureState { + mutable Mutex mutex; + ConditionVariable completed{&mutex}; + bool done ABSL_GUARDED_BY(mutex) = {false}; + ExceptionOr value ABSL_GUARDED_BY(mutex) = + ExceptionOr(Exception::kFailed); + }; + + bool SetExceptionLocked(Exception exception) { + if (!state_->done) { + state_->value = ExceptionOr(exception.value != Exception::kSuccess + ? exception + : Exception{Exception::kFailed}); + state_->done = true; + state_->completed.Notify(); + } + return true; + } + + std::shared_ptr state_ = std::make_shared(); }; } // namespace nearby diff --git a/internal/platform/future_test.cc b/internal/platform/future_test.cc index c7c75d3c..414b36ba 100644 --- a/internal/platform/future_test.cc +++ b/internal/platform/future_test.cc @@ -14,11 +14,11 @@ #include "internal/platform/future.h" +#include + #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" @@ -114,156 +114,4 @@ TEST(FutureTest, GetBlocksWhenNotReady) { EXPECT_GE(blocked_duration, absl::Milliseconds(500)); } -TEST(FutureTest, CallsListenerOnSet) { - constexpr int kValue = 1000; - Future future; - int call_count = 0; - { - SingleThreadExecutor executor; - future.AddListener( - [&](ExceptionOr result) { - ASSERT_TRUE(result.ok()); - ASSERT_EQ(result.GetResult(), kValue); - ++call_count; - }, - &executor); - - future.Set(kValue); - // `executor` leaves scope, the destructor waits for tasks to complete - } - - EXPECT_EQ(call_count, 1); -} - -TEST(FutureTest, CallsAllListenersOnSet) { - constexpr int kValue = 1000; - Future future; - int call_count_listener_1 = 0; - int call_count_listener_2 = 0; - { - SingleThreadExecutor executor; - future.AddListener( - [&](ExceptionOr result) { - ASSERT_TRUE(result.ok()); - ASSERT_EQ(result.GetResult(), kValue); - ++call_count_listener_1; - }, - &executor); - future.AddListener( - [&](ExceptionOr result) { - ASSERT_TRUE(result.ok()); - ASSERT_EQ(result.GetResult(), kValue); - ++call_count_listener_2; - }, - &executor); - - future.Set(kValue); - // `executor` leaves scope, the destructor waits for tasks to complete - } - - EXPECT_EQ(call_count_listener_1, 1); - EXPECT_EQ(call_count_listener_2, 1); -} - -TEST(FutureTest, AddListenerWhenAlreadySetCallsCallback) { - constexpr int kValue = 1000; - Future future; - int call_count = 0; - future.Set(kValue); - { - SingleThreadExecutor executor; - future.AddListener( - [&](ExceptionOr result) { - ASSERT_TRUE(result.ok()); - ASSERT_EQ(result.GetResult(), kValue); - ++call_count; - }, - &executor); - // `executor` leaves scope, the destructor waits for tasks to complete - } - - EXPECT_EQ(call_count, 1); -} - -TEST(FutureTest, CallsListenerOnSetException) { - constexpr Exception kException = {Exception::kFailed}; - Future future; - int call_count = 0; - { - SingleThreadExecutor executor; - future.AddListener( - [&](ExceptionOr result) { - ASSERT_FALSE(result.ok()); - ASSERT_EQ(result.GetException(), kException); - ++call_count; - }, - &executor); - - future.SetException(kException); - // `executor` leaves scope, the destructor waits for tasks to complete - } - - EXPECT_EQ(call_count, 1); -} - -TEST(FutureTest, AddListenerWhenAlreadySetExceptionCallsCallback) { - constexpr Exception kException = {Exception::kFailed}; - Future future; - int call_count = 0; - future.SetException(kException); - { - SingleThreadExecutor executor; - - future.AddListener( - [&](ExceptionOr result) { - ASSERT_FALSE(result.ok()); - ASSERT_EQ(result.GetException(), kException); - ++call_count; - }, - &executor); - - // `executor` leaves scope, the destructor waits for tasks to complete - } - - 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( - [&](ExceptionOr result) { - ASSERT_FALSE(result.ok()); - ASSERT_EQ(result.exception(), Exception::kTimeout); - 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/BUILD b/internal/platform/implementation/BUILD index 3d9a8689..cc0f2e69 100644 --- a/internal/platform/implementation/BUILD +++ b/internal/platform/implementation/BUILD @@ -90,15 +90,12 @@ cc_library( "crypto.h", "device_info.h", "executor.h", - "future.h", "input_file.h", - "listenable_future.h", "log_message.h", "mutex.h", "output_file.h", "preferences_manager.h", "scheduled_executor.h", - "settable_future.h", "submittable_executor.h", "system_clock.h", "timer.h", diff --git a/internal/platform/implementation/future.h b/internal/platform/implementation/future.h deleted file mode 100644 index 837b5911..00000000 --- a/internal/platform/implementation/future.h +++ /dev/null @@ -1,44 +0,0 @@ -// Copyright 2020 Google LLC -// -// Licensed under the Apache License, Version 2.0 (the "License"); -// you may not use this file except in compliance with the License. -// You may obtain a copy of the License at -// -// https://www.apache.org/licenses/LICENSE-2.0 -// -// Unless required by applicable law or agreed to in writing, software -// distributed under the License is distributed on an "AS IS" BASIS, -// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. -// See the License for the specific language governing permissions and -// limitations under the License. - -#ifndef PLATFORM_API_FUTURE_H_ -#define PLATFORM_API_FUTURE_H_ - -#include "absl/time/clock.h" -#include "internal/platform/exception.h" - -namespace nearby { -namespace api { - -// A Future represents the result of an asynchronous computation. -// -// https://docs.oracle.com/javase/8/docs/api/java/util/concurrent/Future.html -template -class Future { - public: - virtual ~Future() = default; - - // throws Exception::kInterrupted, Exception::kExecution - virtual ExceptionOr Get() = 0; - - // throws Exception::kInterrupted, Exception::kExecution - // throws Exception::kTimeout if timeout is exceeded while waiting for - // result. - virtual ExceptionOr Get(absl::Duration timeout) = 0; -}; - -} // namespace api -} // namespace nearby - -#endif // PLATFORM_API_FUTURE_H_ diff --git a/internal/platform/implementation/listenable_future.h b/internal/platform/implementation/listenable_future.h deleted file mode 100644 index 0633c48e..00000000 --- a/internal/platform/implementation/listenable_future.h +++ /dev/null @@ -1,42 +0,0 @@ -// Copyright 2020 Google LLC -// -// Licensed under the Apache License, Version 2.0 (the "License"); -// you may not use this file except in compliance with the License. -// You may obtain a copy of the License at -// -// https://www.apache.org/licenses/LICENSE-2.0 -// -// Unless required by applicable law or agreed to in writing, software -// distributed under the License is distributed on an "AS IS" BASIS, -// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. -// See the License for the specific language governing permissions and -// limitations under the License. - -#ifndef PLATFORM_API_LISTENABLE_FUTURE_H_ -#define PLATFORM_API_LISTENABLE_FUTURE_H_ - -#include - -#include "internal/platform/exception.h" -#include "internal/platform/implementation/executor.h" -#include "internal/platform/implementation/future.h" - -namespace nearby { -namespace api { - -// A Future that accepts completion listeners. -// -// https://guava.dev/releases/20.0/api/docs/com/google/common/util/concurrent/ListenableFuture.html -template -class ListenableFuture : public Future { - public: - using FutureCallback = absl::AnyInvocable)>; - ~ListenableFuture() override = default; - - virtual void AddListener(FutureCallback callback, Executor* executor) = 0; -}; - -} // namespace api -} // namespace nearby - -#endif // PLATFORM_API_LISTENABLE_FUTURE_H_ diff --git a/internal/platform/implementation/platform.h b/internal/platform/implementation/platform.h index 2ddf16b7..6f310750 100644 --- a/internal/platform/implementation/platform.h +++ b/internal/platform/implementation/platform.h @@ -39,7 +39,6 @@ #include "internal/platform/implementation/output_file.h" #include "internal/platform/implementation/scheduled_executor.h" #include "internal/platform/implementation/server_sync.h" -#include "internal/platform/implementation/settable_future.h" #include "internal/platform/implementation/submittable_executor.h" #include "internal/platform/implementation/system_clock.h" #include "internal/platform/implementation/timer.h" @@ -68,7 +67,6 @@ class ImplementationPlatform { // - synchronization primitives: // - mutex (regular, and recursive) // - condition variable (must work with regular mutex only) - // - Future : to synchronize on Callable scheduled to execute. // - CountDownLatch : to ensure at least N threads are waiting. // - file I/O // - Logging diff --git a/internal/platform/implementation/settable_future.h b/internal/platform/implementation/settable_future.h deleted file mode 100644 index beb32f0b..00000000 --- a/internal/platform/implementation/settable_future.h +++ /dev/null @@ -1,47 +0,0 @@ -// Copyright 2020 Google LLC -// -// Licensed under the Apache License, Version 2.0 (the "License"); -// you may not use this file except in compliance with the License. -// You may obtain a copy of the License at -// -// https://www.apache.org/licenses/LICENSE-2.0 -// -// Unless required by applicable law or agreed to in writing, software -// distributed under the License is distributed on an "AS IS" BASIS, -// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. -// See the License for the specific language governing permissions and -// limitations under the License. - -#ifndef PLATFORM_API_SETTABLE_FUTURE_H_ -#define PLATFORM_API_SETTABLE_FUTURE_H_ - -#include "internal/platform/implementation/listenable_future.h" -#include "internal/platform/exception.h" - -namespace nearby { -namespace api { - -// A SettableFuture is a type of Future whose result can be set. -// -// https://google.github.io/guava/releases/20.0/api/docs/com/google/common/util/concurrent/SettableFuture.html -template -class SettableFuture : public ListenableFuture { - public: - ~SettableFuture() override = default; - - // Completes the future successfully. The value is returned to any waiters. - // Returns true, if value was set. - // Returns false, if Future is already in "done" state. - virtual bool Set(T value) = 0; - - // Completes the future unsuccessfully. The exception value is returned to any - // waiters. - // Returns true, if exception was set. - // Returns false, if Future is already in "done" state. - virtual bool SetException(Exception exception) = 0; -}; - -} // namespace api -} // namespace nearby - -#endif // PLATFORM_API_SETTABLE_FUTURE_H_ diff --git a/internal/platform/implementation/submittable_executor.h b/internal/platform/implementation/submittable_executor.h index f523e576..68a23907 100644 --- a/internal/platform/implementation/submittable_executor.h +++ b/internal/platform/implementation/submittable_executor.h @@ -15,10 +15,7 @@ #ifndef PLATFORM_API_SUBMITTABLE_EXECUTOR_H_ #define PLATFORM_API_SUBMITTABLE_EXECUTOR_H_ -#include - #include "internal/platform/implementation/executor.h" -#include "internal/platform/implementation/future.h" #include "internal/platform/runnable.h" namespace nearby { diff --git a/internal/platform/settable_future.h b/internal/platform/settable_future.h deleted file mode 100644 index aeff438a..00000000 --- a/internal/platform/settable_future.h +++ /dev/null @@ -1,168 +0,0 @@ -// Copyright 2020 Google LLC -// -// Licensed under the Apache License, Version 2.0 (the "License"); -// you may not use this file except in compliance with the License. -// You may obtain a copy of the License at -// -// https://www.apache.org/licenses/LICENSE-2.0 -// -// Unless required by applicable law or agreed to in writing, software -// distributed under the License is distributed on an "AS IS" BASIS, -// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. -// See the License for the specific language governing permissions and -// limitations under the License. - -#ifndef PLATFORM_PUBLIC_SETTABLE_FUTURE_H_ -#define PLATFORM_PUBLIC_SETTABLE_FUTURE_H_ - -#include -#include -#include - -#include "absl/time/time.h" -#include "internal/flags/nearby_flags.h" -#include "internal/platform/condition_variable.h" -#include "internal/platform/exception.h" -#include "internal/platform/flags/nearby_platform_feature_flags.h" -#include "internal/platform/implementation/executor.h" -#include "internal/platform/implementation/listenable_future.h" -#include "internal/platform/implementation/platform.h" -#include "internal/platform/implementation/settable_future.h" -#include "internal/platform/implementation/submittable_executor.h" -#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 { - -template -class SettableFuture : public api::SettableFuture { - public: - using FutureCallback = typename api::ListenableFuture::FutureCallback; - SettableFuture() = default; - - // Creates a SettableFuture that fails with a kTimeout when `timeout` expires. - explicit SettableFuture(absl::Duration timeout) - : timer_(std::make_unique()) { - timer_->Start(absl::ToInt64Milliseconds(timeout), 0, [this] { - // Offload the timeout to a single thread executor. - executor_ = api::ImplementationPlatform::CreateSingleThreadExecutor(); - executor_->Execute([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; - exception_ = {Exception::kSuccess}; - completed_.Notify(); - InvokeAllLocked(); - return true; - } - return false; - } - - void AddListener(FutureCallback callback, api::Executor* executor) override { - MutexLock lock(&mutex_); - if (done_) { - executor->Execute( - [value = GetLocked(), callback = std::move(callback)]() mutable { - callback(std::move(value)); - }); - } else { - listeners_.emplace_back(std::make_pair(executor, std::move(callback))); - } - } - - bool IsSet() const { - MutexLock lock(&mutex_); - return done_; - } - - 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); - } - - ExceptionOr Get() override { - MutexLock lock(&mutex_); - while (!done_) { - completed_.Wait(); - } - return GetLocked(); - } - - ExceptionOr Get(absl::Duration timeout) override { - MutexLock lock(&mutex_); - while (!done_) { - absl::Time start_time = SystemClock::ElapsedRealtime(); - if (completed_.Wait(timeout).Raised(Exception::kInterrupted)) { - SetExceptionLocked({Exception::kInterrupted}); - break; - } - absl::Duration spent = SystemClock::ElapsedRealtime() - start_time; - if (spent < timeout) { - timeout -= spent; - } else if (!done_) { - SetExceptionLocked({Exception::kTimeout}); - break; - } - } - return GetLocked(); - } - - private: - bool SetExceptionLocked(Exception exception) { - if (!done_) { - exception_ = exception.value != Exception::kSuccess - ? exception - : Exception{Exception::kFailed}; - done_ = true; - completed_.Notify(); - InvokeAllLocked(); - } - return true; - } - - ExceptionOr GetLocked() { - return exception_.value != Exception::kSuccess - ? ExceptionOr{exception_.value} - : ExceptionOr{value_}; - } - - void InvokeAllLocked() { - for (auto& item : listeners_) { - item.first->Execute( - [value = GetLocked(), callback = std::move(item.second)]() mutable { - callback(std::move(value)); - }); - } - listeners_.clear(); - } - - mutable Mutex mutex_; - ConditionVariable completed_{&mutex_}; - std::vector> listeners_; - bool done_{false}; - T value_; - Exception exception_{Exception::kFailed}; - std::unique_ptr timer_; - std::unique_ptr executor_; -}; - -} // namespace nearby - -#endif // PLATFORM_PUBLIC_SETTABLE_FUTURE_H_