TySS-Dev/nearby
1
0
Fork 0
mirror of https://github.com/kidfromjupiter/nearby.git synced 2026-09-14 14:46:12 -04:00
Code Issues Packages Projects Releases Wiki Activity

Internal cleanup

Browse Source 
PiperOrigin-RevId: 817215477
This commit is contained in:
Guogang Li
2025-10-09 09:17:07 -07:00
committed by Copybara-Service
parent 7145975cd4
commit 9eca355226
10 changed files with 73 additions and 492 deletions
Download Patch File Download Diff File Expand all files Collapse all files
internal/platform/BUILD
-1
View File
@@ -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",
internal/platform/future.h
+71 -28
View File
@@ -18,49 +18,92 @@
#include <memory>
#include <utility>
#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 <typename T>
class Future final {
public:
using FutureCallback = typename SettableFuture<T>::FutureCallback;
// Default Future. Does not time out.
Future() : impl_(std::make_shared<SettableFuture<T>>()) {}
// Sets the value of the Future.
bool Set(T value) {
MutexLock lock(&state_->mutex);
if (!state_->done) {
state_->value = ExceptionOr<T>(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<SettableFuture<T>>(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<T> Get() {
MutexLock lock(&state_->mutex);
while (!state_->done) {
state_->completed.Wait();
}
return state_->value;
}
virtual ExceptionOr<T> Get() { return impl_->Get(); }
virtual ExceptionOr<T> Get(absl::Duration timeout) {
return impl_->Get(timeout);
// Gets the value of the Future, timing out after the specified duration.
ExceptionOr<T> 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<bool> future;
// RunOnXyzThread([future]() { future.Set(DoTheJobAndReport()); });
// if (future.Get().Ok()) { /*...*/ }
// 2)
// Future<bool> future = DoSomeAsyncWork(); // Returns future, but keeps copy.
// if (future.Get().Ok()) { /*...*/ }
std::shared_ptr<SettableFuture<T>> impl_;
struct FutureState {
mutable Mutex mutex;
ConditionVariable completed{&mutex};
bool done ABSL_GUARDED_BY(mutex) = {false};
ExceptionOr<T> value ABSL_GUARDED_BY(mutex) =
ExceptionOr<T>(Exception::kFailed);
};
bool SetExceptionLocked(Exception exception) {
if (!state_->done) {
state_->value = ExceptionOr<T>(exception.value != Exception::kSuccess
? exception
: Exception{Exception::kFailed});
state_->done = true;
state_->completed.Notify();
}
return true;
}
std::shared_ptr<FutureState> state_ = std::make_shared<FutureState>();
};
} // namespace nearby
internal/platform/future_test.cc
+2 -154
View File
@@ -14,11 +14,11 @@
#include "internal/platform/future.h"
#include <cstring>
#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<int> future;
int call_count = 0;
{
SingleThreadExecutor executor;
future.AddListener(
[&](ExceptionOr<int> 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<int> future;
int call_count_listener_1 = 0;
int call_count_listener_2 = 0;
{
SingleThreadExecutor executor;
future.AddListener(
[&](ExceptionOr<int> result) {
ASSERT_TRUE(result.ok());
ASSERT_EQ(result.GetResult(), kValue);
++call_count_listener_1;
},
&executor);
future.AddListener(
[&](ExceptionOr<int> 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<int> future;
int call_count = 0;
future.Set(kValue);
{
SingleThreadExecutor executor;
future.AddListener(
[&](ExceptionOr<int> 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<int> future;
int call_count = 0;
{
SingleThreadExecutor executor;
future.AddListener(
[&](ExceptionOr<int> 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<int> future;
int call_count = 0;
future.SetException(kException);
{
SingleThreadExecutor executor;
future.AddListener(
[&](ExceptionOr<int> 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<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(
[&](ExceptionOr<int> 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<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/BUILD
-3
View File
@@ -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",
internal/platform/implementation/future.h
-44
View File
@@ -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 <typename T>
class Future {
public:
virtual ~Future() = default;
// throws Exception::kInterrupted, Exception::kExecution
virtual ExceptionOr<T> Get() = 0;
// throws Exception::kInterrupted, Exception::kExecution
// throws Exception::kTimeout if timeout is exceeded while waiting for
// result.
virtual ExceptionOr<T> Get(absl::Duration timeout) = 0;
};
} // namespace api
} // namespace nearby
#endif // PLATFORM_API_FUTURE_H_
internal/platform/implementation/listenable_future.h
-42
View File
@@ -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 <memory>
#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 <typename T>
class ListenableFuture : public Future<T> {
public:
using FutureCallback = absl::AnyInvocable<void(ExceptionOr<T>)>;
~ListenableFuture() override = default;
virtual void AddListener(FutureCallback callback, Executor* executor) = 0;
};
} // namespace api
} // namespace nearby
#endif // PLATFORM_API_LISTENABLE_FUTURE_H_
internal/platform/implementation/platform.h
-2
View File
@@ -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<T> : to synchronize on Callable<T> scheduled to execute.
// - CountDownLatch : to ensure at least N threads are waiting.
// - file I/O
// - Logging
internal/platform/implementation/settable_future.h
-47
View File
@@ -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 <typename T>
class SettableFuture : public ListenableFuture<T> {
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_
internal/platform/implementation/submittable_executor.h
-3
View File
@@ -15,10 +15,7 @@
#ifndef PLATFORM_API_SUBMITTABLE_EXECUTOR_H_
#define PLATFORM_API_SUBMITTABLE_EXECUTOR_H_
#include <memory>
#include "internal/platform/implementation/executor.h"
#include "internal/platform/implementation/future.h"
#include "internal/platform/runnable.h"
namespace nearby {
internal/platform/settable_future.h
-168
View File
@@ -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 <memory>
#include <utility>
#include <vector>
#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 <typename T>
class SettableFuture : public api::SettableFuture<T> {
public:
using FutureCallback = typename api::ListenableFuture<T>::FutureCallback;
SettableFuture() = default;
// Creates a SettableFuture that fails with a kTimeout when `timeout` expires.
explicit SettableFuture(absl::Duration timeout)
: timer_(std::make_unique<TimerImpl>()) {
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<T> Get() override {
MutexLock lock(&mutex_);
while (!done_) {
completed_.Wait();
}
return GetLocked();
}
ExceptionOr<T> 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<T> GetLocked() {
return exception_.value != Exception::kSuccess
? ExceptionOr<T>{exception_.value}
: ExceptionOr<T>{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<std::pair<api::Executor*, FutureCallback>> listeners_;
bool done_{false};
T value_;
Exception exception_{Exception::kFailed};
std::unique_ptr<TimerImpl> timer_;
std::unique_ptr<api::SubmittableExecutor> executor_;
};
} // namespace nearby
#endif // PLATFORM_PUBLIC_SETTABLE_FUTURE_H_