mirror of
https://github.com/kidfromjupiter/nearby.git
synced 2026-09-15 15:16:12 -04:00
Fixed some test issues
PiperOrigin-RevId: 662722512
This commit is contained in:
Guogang Li
committed by
Copybara-Service
Copybara-Service
parent
cad1ef6bf3
commit
25eec0ff0f
@@ -359,6 +359,8 @@ cc_library(
|
||||
":util",
|
||||
"//internal/base:files",
|
||||
"//internal/crypto_cros",
|
||||
"//internal/flags:nearby_flags",
|
||||
"//internal/platform/flags:platform_flags",
|
||||
"//internal/platform/implementation:platform",
|
||||
"//internal/platform/implementation:types",
|
||||
"@com_google_absl//absl/base:core_headers",
|
||||
|
||||
@@ -15,8 +15,12 @@
|
||||
#ifndef PLATFORM_PUBLIC_FUTURE_H_
|
||||
#define PLATFORM_PUBLIC_FUTURE_H_
|
||||
|
||||
#include <memory>
|
||||
#include <utility>
|
||||
|
||||
#include "absl/time/time.h"
|
||||
#include "internal/platform/exception.h"
|
||||
#include "internal/platform/implementation/executor.h"
|
||||
#include "internal/platform/settable_future.h"
|
||||
|
||||
namespace nearby {
|
||||
|
||||
@@ -78,7 +78,7 @@ TEST(FutureTest, SupportScopedEnum) {
|
||||
}
|
||||
|
||||
TEST(FutureTest, SetTakesCopyOfValue) {
|
||||
// Default constructor is zero-initalizing all data in BigSizedStruct.
|
||||
// Default constructor is zero-initializing all data in BigSizedStruct.
|
||||
BigSizedStruct v1;
|
||||
Future<BigSizedStruct> future;
|
||||
v1.data[0] = 5; // Changing value before calling Set() will affect stored
|
||||
|
||||
@@ -19,7 +19,6 @@
|
||||
#include <memory>
|
||||
#include <utility>
|
||||
|
||||
#include "absl/synchronization/mutex.h"
|
||||
#include "absl/time/time.h"
|
||||
#include "internal/flags/nearby_flags.h"
|
||||
#include "internal/platform/flags/nearby_platform_feature_flags.h"
|
||||
@@ -32,7 +31,10 @@ namespace windows {
|
||||
|
||||
ScheduledExecutor::ScheduledExecutor()
|
||||
: executor_(std::make_unique<nearby::windows::Executor>()),
|
||||
shut_down_(false) {}
|
||||
shut_down_(false),
|
||||
use_task_scheduler_(NearbyFlags::GetInstance().GetBoolFlag(
|
||||
platform::config_package_nearby::nearby_platform_feature::
|
||||
kEnableTaskScheduler)) {}
|
||||
|
||||
// Cancelable is kept both in the executor context, and in the caller context.
|
||||
// We want Cancelable to live until both caller and executor are done with it.
|
||||
@@ -40,10 +42,13 @@ ScheduledExecutor::ScheduledExecutor()
|
||||
// using std:shared_ptr<> instead of std::unique_ptr<>.
|
||||
std::shared_ptr<api::Cancelable> ScheduledExecutor::Schedule(
|
||||
Runnable&& runnable, absl::Duration duration) {
|
||||
absl::MutexLock lock(&mutex_);
|
||||
if (NearbyFlags::GetInstance().GetBoolFlag(
|
||||
platform::config_package_nearby::nearby_platform_feature::
|
||||
kEnableTaskScheduler)) {
|
||||
if (use_task_scheduler_) {
|
||||
if (shut_down_) {
|
||||
NEARBY_LOGS(ERROR) << __func__
|
||||
<< ": Attempt to Schedule on a shut down executor.";
|
||||
|
||||
return nullptr;
|
||||
}
|
||||
return task_scheduler_.Schedule(std::move(runnable), duration);
|
||||
} else {
|
||||
if (shut_down_) {
|
||||
@@ -73,7 +78,6 @@ std::shared_ptr<api::Cancelable> ScheduledExecutor::Schedule(
|
||||
}
|
||||
|
||||
void ScheduledExecutor::Execute(Runnable&& runnable) {
|
||||
absl::MutexLock lock(&mutex_);
|
||||
if (shut_down_) {
|
||||
NEARBY_LOGS(ERROR) << __func__
|
||||
<< ": Attempt to Execute on a shut down executor.";
|
||||
@@ -84,16 +88,24 @@ void ScheduledExecutor::Execute(Runnable&& runnable) {
|
||||
}
|
||||
|
||||
void ScheduledExecutor::Shutdown() {
|
||||
absl::MutexLock lock(&mutex_);
|
||||
if (!shut_down_) {
|
||||
shut_down_ = true;
|
||||
for (auto& task : scheduled_tasks_) {
|
||||
task->Cancel();
|
||||
if (use_task_scheduler_) {
|
||||
if (!shut_down_) {
|
||||
shut_down_ = true;
|
||||
executor_->Shutdown();
|
||||
task_scheduler_.Shutdown();
|
||||
return;
|
||||
}
|
||||
} else {
|
||||
if (!shut_down_) {
|
||||
shut_down_ = true;
|
||||
for (auto& task : scheduled_tasks_) {
|
||||
task->Cancel();
|
||||
}
|
||||
|
||||
scheduled_tasks_.clear();
|
||||
executor_->Shutdown();
|
||||
return;
|
||||
scheduled_tasks_.clear();
|
||||
executor_->Shutdown();
|
||||
return;
|
||||
}
|
||||
}
|
||||
NEARBY_LOGS(ERROR) << __func__
|
||||
<< ": Attempt to Shutdown on a shut down executor.";
|
||||
|
||||
@@ -19,11 +19,8 @@
|
||||
|
||||
#include <atomic>
|
||||
#include <memory>
|
||||
#include <utility>
|
||||
#include <vector>
|
||||
|
||||
#include "absl/base/thread_annotations.h"
|
||||
#include "absl/synchronization/mutex.h"
|
||||
#include "absl/synchronization/notification.h"
|
||||
#include "absl/time/time.h"
|
||||
#include "internal/platform/implementation/cancelable.h"
|
||||
@@ -52,14 +49,13 @@ class ScheduledExecutor : public api::ScheduledExecutor {
|
||||
// Exclusive ownership model does not work for this case;
|
||||
// using std:shared_ptr<> instead if std::unique_ptr<>.
|
||||
std::shared_ptr<api::Cancelable> Schedule(Runnable&& runnable,
|
||||
absl::Duration duration) override
|
||||
ABSL_LOCKS_EXCLUDED(mutex_);
|
||||
absl::Duration duration) override;
|
||||
|
||||
// Executes the runnable task immediately.
|
||||
void Execute(Runnable&& runnable) override ABSL_LOCKS_EXCLUDED(mutex_);
|
||||
void Execute(Runnable&& runnable) override;
|
||||
|
||||
// Shutdowns the executor, all scheduled task will be cancelled.
|
||||
void Shutdown() override ABSL_LOCKS_EXCLUDED(mutex_);
|
||||
void Shutdown() override;
|
||||
|
||||
private:
|
||||
class ScheduledTask : public api::Cancelable {
|
||||
@@ -97,13 +93,12 @@ class ScheduledExecutor : public api::ScheduledExecutor {
|
||||
bool is_executed_ = false;
|
||||
};
|
||||
|
||||
absl::Mutex mutex_;
|
||||
std::unique_ptr<nearby::windows::Executor> executor_ ABSL_GUARDED_BY(mutex_) =
|
||||
nullptr;
|
||||
std::vector<std::shared_ptr<ScheduledTask>> scheduled_tasks_
|
||||
ABSL_GUARDED_BY(mutex_);
|
||||
std::atomic_bool shut_down_ ABSL_GUARDED_BY(mutex_) = false;
|
||||
TaskScheduler task_scheduler_ ABSL_GUARDED_BY(mutex_);
|
||||
std::unique_ptr<nearby::windows::Executor> executor_ = nullptr;
|
||||
std::vector<std::shared_ptr<ScheduledTask>> scheduled_tasks_;
|
||||
std::atomic_bool shut_down_ = false;
|
||||
|
||||
const bool use_task_scheduler_;
|
||||
TaskScheduler task_scheduler_;
|
||||
};
|
||||
|
||||
} // namespace windows
|
||||
|
||||
@@ -17,7 +17,6 @@
|
||||
#include <chrono> // NOLINT
|
||||
#include <memory>
|
||||
#include <utility>
|
||||
#include <vector>
|
||||
|
||||
#include "gtest/gtest.h"
|
||||
#include "absl/synchronization/notification.h"
|
||||
@@ -31,8 +30,9 @@ namespace nearby {
|
||||
namespace windows {
|
||||
namespace {
|
||||
|
||||
class ScheduledExecutorTaskSchedulerFlagTest
|
||||
: public ::testing::TestWithParam<bool> {
|
||||
constexpr absl::Duration kWaitTimeout = absl::Milliseconds(2000);
|
||||
|
||||
class ScheduledExecutorTest : public ::testing::TestWithParam<bool> {
|
||||
public:
|
||||
void SetUp() override {
|
||||
NearbyFlags::GetInstance().OverrideBoolFlagValue(
|
||||
@@ -40,9 +40,13 @@ class ScheduledExecutorTaskSchedulerFlagTest
|
||||
kEnableTaskScheduler,
|
||||
GetParam());
|
||||
}
|
||||
|
||||
void TearDown() override {
|
||||
NearbyFlags::GetInstance().ResetOverridedValues();
|
||||
}
|
||||
};
|
||||
|
||||
TEST_P(ScheduledExecutorTaskSchedulerFlagTest, ExecuteSucceeds) {
|
||||
TEST_P(ScheduledExecutorTest, ExecuteSucceeds) {
|
||||
absl::Notification notification;
|
||||
// Arrange
|
||||
std::string expected(RUNNABLE_0_TEXT.c_str());
|
||||
@@ -62,8 +66,7 @@ TEST_P(ScheduledExecutorTaskSchedulerFlagTest, ExecuteSucceeds) {
|
||||
notification.Notify();
|
||||
});
|
||||
|
||||
ASSERT_TRUE(
|
||||
notification.WaitForNotificationWithTimeout(absl::Milliseconds(200)));
|
||||
ASSERT_TRUE(notification.WaitForNotificationWithTimeout(kWaitTimeout));
|
||||
submittableExecutor->Shutdown();
|
||||
|
||||
// Assert
|
||||
@@ -76,7 +79,7 @@ TEST_P(ScheduledExecutorTaskSchedulerFlagTest, ExecuteSucceeds) {
|
||||
ASSERT_EQ(output, expected);
|
||||
}
|
||||
|
||||
TEST_P(ScheduledExecutorTaskSchedulerFlagTest, ScheduleSucceeds) {
|
||||
TEST_P(ScheduledExecutorTest, ScheduleSucceeds) {
|
||||
absl::Notification notification;
|
||||
// Arrange
|
||||
std::string expected(RUNNABLE_0_TEXT.c_str());
|
||||
@@ -103,8 +106,7 @@ TEST_P(ScheduledExecutorTaskSchedulerFlagTest, ScheduleSucceeds) {
|
||||
},
|
||||
absl::Milliseconds(50));
|
||||
|
||||
ASSERT_TRUE(
|
||||
notification.WaitForNotificationWithTimeout(absl::Milliseconds(200)));
|
||||
ASSERT_TRUE(notification.WaitForNotificationWithTimeout(kWaitTimeout));
|
||||
submittableExecutor->Shutdown();
|
||||
|
||||
ASSERT_EQ(threadIds->size(), 2);
|
||||
@@ -114,7 +116,7 @@ TEST_P(ScheduledExecutorTaskSchedulerFlagTest, ScheduleSucceeds) {
|
||||
ASSERT_EQ(output, expected);
|
||||
}
|
||||
|
||||
TEST_P(ScheduledExecutorTaskSchedulerFlagTest, CancelSucceeds) {
|
||||
TEST_P(ScheduledExecutorTest, CancelSucceeds) {
|
||||
absl::Notification notification;
|
||||
// Arrange
|
||||
std::string expected("");
|
||||
@@ -138,8 +140,7 @@ TEST_P(ScheduledExecutorTaskSchedulerFlagTest, CancelSucceeds) {
|
||||
|
||||
auto actual = cancelable->Cancel();
|
||||
|
||||
EXPECT_FALSE(
|
||||
notification.WaitForNotificationWithTimeout(absl::Milliseconds(2000)));
|
||||
EXPECT_FALSE(notification.WaitForNotificationWithTimeout(kWaitTimeout));
|
||||
submittableExecutor->Shutdown();
|
||||
|
||||
// Assert
|
||||
@@ -151,7 +152,7 @@ TEST_P(ScheduledExecutorTaskSchedulerFlagTest, CancelSucceeds) {
|
||||
ASSERT_EQ(output, expected);
|
||||
}
|
||||
|
||||
TEST_P(ScheduledExecutorTaskSchedulerFlagTest, CancelAfterStartedFails) {
|
||||
TEST_P(ScheduledExecutorTest, CancelAfterStartedFails) {
|
||||
absl::Notification notification;
|
||||
// Arrange
|
||||
std::string expected(RUNNABLE_0_TEXT.c_str());
|
||||
@@ -173,21 +174,14 @@ TEST_P(ScheduledExecutorTaskSchedulerFlagTest, CancelAfterStartedFails) {
|
||||
},
|
||||
absl::Milliseconds(100));
|
||||
|
||||
absl::SleepFor(absl::Milliseconds(200));
|
||||
absl::SleepFor(absl::Milliseconds(500));
|
||||
auto actual = cancelable->Cancel();
|
||||
|
||||
ASSERT_TRUE(
|
||||
notification.WaitForNotificationWithTimeout(absl::Milliseconds(2000)));
|
||||
ASSERT_TRUE(notification.WaitForNotificationWithTimeout(kWaitTimeout));
|
||||
submittableExecutor->Shutdown();
|
||||
|
||||
// Assert
|
||||
if (NearbyFlags::GetInstance().GetBoolFlag(
|
||||
platform::config_package_nearby::nearby_platform_feature::
|
||||
kEnableTaskScheduler)) {
|
||||
ASSERT_TRUE(actual);
|
||||
} else {
|
||||
ASSERT_FALSE(actual);
|
||||
}
|
||||
ASSERT_FALSE(actual);
|
||||
ASSERT_EQ(threadIds->size(), 2);
|
||||
// We should still be on the main thread
|
||||
ASSERT_EQ(GetCurrentThreadId(), threadIds->at(0));
|
||||
@@ -195,9 +189,8 @@ TEST_P(ScheduledExecutorTaskSchedulerFlagTest, CancelAfterStartedFails) {
|
||||
ASSERT_EQ(output, expected);
|
||||
}
|
||||
|
||||
INSTANTIATE_TEST_SUITE_P(ScheduledExecutorTest,
|
||||
ScheduledExecutorTaskSchedulerFlagTest,
|
||||
testing::ValuesIn(std::vector<bool>{true, false}));
|
||||
INSTANTIATE_TEST_SUITE_P(ScheduledExecutorTaskSchedulerFlagTest,
|
||||
ScheduledExecutorTest, testing::Bool());
|
||||
|
||||
} // namespace
|
||||
} // namespace windows
|
||||
|
||||
@@ -39,9 +39,9 @@ void CALLBACK TimerRoutine(PVOID lpParam, BOOLEAN TimerOrWaitFired) {
|
||||
TaskScheduler::TaskScheduler() {
|
||||
NEARBY_LOGS(INFO) << __func__ << ": Created task scheduler: " << this;
|
||||
}
|
||||
|
||||
TaskScheduler::~TaskScheduler() {
|
||||
absl::MutexLock lock(&mutex_);
|
||||
ShutdownInternal();
|
||||
Shutdown();
|
||||
NEARBY_LOGS(INFO) << __func__ << ": Destroyed task scheduler: " << this;
|
||||
}
|
||||
|
||||
@@ -59,9 +59,19 @@ std::shared_ptr<api::Cancelable> TaskScheduler::Schedule(
|
||||
<< ", duration: " << absl::ToInt64Milliseconds(duration)
|
||||
<< "ms, repeat_interval: "
|
||||
<< absl::ToInt64Milliseconds(repeat_interval) << "ms";
|
||||
if (is_shutdown_) {
|
||||
NEARBY_LOGS(ERROR) << __func__
|
||||
<< ": Attempt to schedule task on a shut down task "
|
||||
"scheduler: "
|
||||
<< this;
|
||||
return nullptr;
|
||||
}
|
||||
|
||||
std::shared_ptr<ScheduledTask> task =
|
||||
std::make_shared<ScheduledTask>(*this, std::move(runnable));
|
||||
// Clear all cancelled tasks.
|
||||
CleanScheduledTasks();
|
||||
|
||||
std::shared_ptr<ScheduledTask> task = std::make_shared<ScheduledTask>(
|
||||
*this, std::move(runnable), repeat_interval != absl::ZeroDuration());
|
||||
|
||||
HANDLE timer_handle = nullptr;
|
||||
if (!CreateTimerQueueTimer(&timer_handle, nullptr,
|
||||
@@ -76,7 +86,7 @@ std::shared_ptr<api::Cancelable> TaskScheduler::Schedule(
|
||||
return nullptr;
|
||||
}
|
||||
|
||||
task->SetTimerHandle(reinterpret_cast<intptr_t>(timer_handle));
|
||||
task->set_timer_handle(reinterpret_cast<intptr_t>(timer_handle));
|
||||
scheduled_tasks_.insert({reinterpret_cast<intptr_t>(timer_handle), task});
|
||||
NEARBY_LOGS(INFO) << __func__ << ": Scheduled task " << task.get()
|
||||
<< " on task scheduler:" << this
|
||||
@@ -87,29 +97,87 @@ std::shared_ptr<api::Cancelable> TaskScheduler::Schedule(
|
||||
void TaskScheduler::Shutdown() {
|
||||
absl::MutexLock lock(&mutex_);
|
||||
NEARBY_LOGS(INFO) << __func__ << ": Shutting down task scheduler:" << this;
|
||||
ShutdownInternal();
|
||||
if (is_shutdown_) {
|
||||
return;
|
||||
}
|
||||
for (auto& task : scheduled_tasks_) {
|
||||
if (task.second->is_cancelled()) {
|
||||
continue;
|
||||
}
|
||||
// Wait for running task to finish.
|
||||
if (!DeleteTimerQueueTimer(
|
||||
nullptr, reinterpret_cast<HANDLE>(task.second->timer_handle()),
|
||||
INVALID_HANDLE_VALUE)) {
|
||||
if (GetLastError() != ERROR_IO_PENDING) {
|
||||
NEARBY_LOGS(ERROR) << __func__
|
||||
<< ": Failed to delete timer queue timer: "
|
||||
<< task.second->timer_handle()
|
||||
<< " error: " << GetLastError();
|
||||
}
|
||||
}
|
||||
}
|
||||
scheduled_tasks_.clear();
|
||||
is_shutdown_ = true;
|
||||
NEARBY_LOGS(INFO) << __func__ << ": Shut down task scheduler:" << this;
|
||||
}
|
||||
|
||||
TaskScheduler::ScheduledTask::ScheduledTask(TaskScheduler& task_scheduler,
|
||||
Runnable&& runnable)
|
||||
: task_scheduler_(&task_scheduler), runnable_(std::move(runnable)) {}
|
||||
|
||||
bool TaskScheduler::ScheduledTask::Cancel() {
|
||||
NEARBY_LOGS(INFO) << __func__ << ": Cancelling timer " << timer_handle_
|
||||
<< " from task scheduler:" << this;
|
||||
return task_scheduler_->RemoveScheduledTask(timer_handle_);
|
||||
Runnable&& runnable,
|
||||
bool is_repeated)
|
||||
: task_scheduler_(&task_scheduler), is_repeated_(is_repeated) {
|
||||
runnable_ = [this, runnable = std::move(runnable)]() mutable {
|
||||
{
|
||||
absl::MutexLock lock(&mutex_);
|
||||
is_executed_ = true;
|
||||
}
|
||||
if (runnable) {
|
||||
runnable();
|
||||
}
|
||||
};
|
||||
}
|
||||
|
||||
void TaskScheduler::ScheduledTask::SetTimerHandle(intptr_t timer_handle) {
|
||||
bool TaskScheduler::ScheduledTask::Cancel() {
|
||||
NEARBY_LOGS(INFO) << __func__ << ": Cancelling timer " << timer_handle()
|
||||
<< " from task scheduler:" << this;
|
||||
{
|
||||
absl::MutexLock lock(&mutex_);
|
||||
if (is_cancelled_) {
|
||||
return false;
|
||||
}
|
||||
is_cancelled_ = true;
|
||||
}
|
||||
|
||||
bool result = task_scheduler_->CancelScheduledTask(timer_handle());
|
||||
{
|
||||
absl::MutexLock lock(&mutex_);
|
||||
if (!is_repeated_ && is_executed_) {
|
||||
result = false;
|
||||
}
|
||||
}
|
||||
return result;
|
||||
}
|
||||
|
||||
void TaskScheduler::ScheduledTask::set_timer_handle(intptr_t timer_handle) {
|
||||
absl::MutexLock lock(&mutex_);
|
||||
timer_handle_ = timer_handle;
|
||||
}
|
||||
|
||||
Runnable* TaskScheduler::ScheduledTask::runnable() { return &runnable_; }
|
||||
Runnable* TaskScheduler::ScheduledTask::runnable() {
|
||||
absl::MutexLock lock(&mutex_);
|
||||
return &runnable_;
|
||||
}
|
||||
|
||||
intptr_t TaskScheduler::ScheduledTask::timer_handle() { return timer_handle_; }
|
||||
intptr_t TaskScheduler::ScheduledTask::timer_handle() const {
|
||||
absl::MutexLock lock(&mutex_);
|
||||
return timer_handle_;
|
||||
}
|
||||
|
||||
bool TaskScheduler::RemoveScheduledTask(intptr_t timer_handle) {
|
||||
bool TaskScheduler::ScheduledTask::is_cancelled() const {
|
||||
absl::MutexLock lock(&mutex_);
|
||||
return is_cancelled_;
|
||||
}
|
||||
|
||||
bool TaskScheduler::CancelScheduledTask(intptr_t timer_handle) {
|
||||
absl::MutexLock lock(&mutex_);
|
||||
auto it = scheduled_tasks_.find(timer_handle);
|
||||
if (it == scheduled_tasks_.end()) {
|
||||
@@ -126,29 +194,18 @@ bool TaskScheduler::RemoveScheduledTask(intptr_t timer_handle) {
|
||||
}
|
||||
}
|
||||
|
||||
scheduled_tasks_.erase(it);
|
||||
return true;
|
||||
}
|
||||
|
||||
void TaskScheduler::ShutdownInternal() {
|
||||
if (is_shutdown_) {
|
||||
return;
|
||||
}
|
||||
for (auto& task : scheduled_tasks_) {
|
||||
// Wait for running task to finish.
|
||||
if (!DeleteTimerQueueTimer(
|
||||
nullptr, reinterpret_cast<HANDLE>(task.second->timer_handle()),
|
||||
INVALID_HANDLE_VALUE)) {
|
||||
if (GetLastError() != ERROR_IO_PENDING) {
|
||||
NEARBY_LOGS(ERROR) << __func__
|
||||
<< ": Failed to delete timer queue timer: "
|
||||
<< task.second->timer_handle()
|
||||
<< " error: " << GetLastError();
|
||||
}
|
||||
void TaskScheduler::CleanScheduledTasks() {
|
||||
auto it = scheduled_tasks_.begin();
|
||||
while (it != scheduled_tasks_.end()) {
|
||||
if (it->second->is_cancelled()) {
|
||||
scheduled_tasks_.erase(it++);
|
||||
} else {
|
||||
++it;
|
||||
}
|
||||
}
|
||||
scheduled_tasks_.clear();
|
||||
is_shutdown_ = true;
|
||||
}
|
||||
|
||||
} // namespace nearby::windows
|
||||
|
||||
@@ -47,26 +47,32 @@ class TaskScheduler {
|
||||
private:
|
||||
class ScheduledTask : public api::Cancelable {
|
||||
public:
|
||||
explicit ScheduledTask(TaskScheduler& task_scheduler, Runnable&& runnable);
|
||||
explicit ScheduledTask(TaskScheduler& task_scheduler, Runnable&& runnable,
|
||||
bool is_repeated);
|
||||
~ScheduledTask() override = default;
|
||||
|
||||
// Note: not support to cancel and shutdown a scheduled task in the
|
||||
// callback.
|
||||
bool Cancel() override;
|
||||
// when task is cancelled or executed, return false, otherwise return true.
|
||||
bool Cancel() override ABSL_LOCKS_EXCLUDED(mutex_);
|
||||
bool is_cancelled() const ABSL_LOCKS_EXCLUDED(mutex_);
|
||||
|
||||
void SetTimerHandle(intptr_t timer_handle);
|
||||
intptr_t timer_handle();
|
||||
|
||||
Runnable* runnable();
|
||||
intptr_t timer_handle() const ABSL_LOCKS_EXCLUDED(mutex_);
|
||||
void set_timer_handle(intptr_t timer_handle) ABSL_LOCKS_EXCLUDED(mutex_);
|
||||
Runnable* runnable() ABSL_LOCKS_EXCLUDED(mutex_);
|
||||
|
||||
private:
|
||||
TaskScheduler* task_scheduler_;
|
||||
Runnable runnable_;
|
||||
intptr_t timer_handle_;
|
||||
mutable absl::Mutex mutex_;
|
||||
TaskScheduler* const task_scheduler_;
|
||||
Runnable runnable_ ABSL_GUARDED_BY(mutex_);
|
||||
intptr_t timer_handle_ ABSL_GUARDED_BY(mutex_);
|
||||
bool is_cancelled_ ABSL_GUARDED_BY(mutex_) = false;
|
||||
bool is_executed_ ABSL_GUARDED_BY(mutex_) = false;
|
||||
bool is_repeated_ ABSL_GUARDED_BY(mutex_) = false;
|
||||
};
|
||||
|
||||
bool RemoveScheduledTask(intptr_t timer_handle) ABSL_LOCKS_EXCLUDED(mutex_);
|
||||
void ShutdownInternal() ABSL_EXCLUSIVE_LOCKS_REQUIRED(mutex_);
|
||||
bool CancelScheduledTask(intptr_t timer_handle) ABSL_LOCKS_EXCLUDED(mutex_);
|
||||
void CleanScheduledTasks() ABSL_EXCLUSIVE_LOCKS_REQUIRED(mutex_);
|
||||
|
||||
absl::Mutex mutex_;
|
||||
bool is_shutdown_ ABSL_GUARDED_BY(mutex_) = false;
|
||||
|
||||
@@ -29,13 +29,16 @@
|
||||
namespace nearby {
|
||||
namespace windows {
|
||||
|
||||
Timer::Timer()
|
||||
: use_task_scheduler_(NearbyFlags::GetInstance().GetBoolFlag(
|
||||
platform::config_package_nearby::nearby_platform_feature::
|
||||
kEnableTaskScheduler)) {}
|
||||
|
||||
Timer::~Timer() { Stop(); }
|
||||
|
||||
bool Timer::Create(int delay, int interval,
|
||||
absl::AnyInvocable<void()> callback) {
|
||||
if (NearbyFlags::GetInstance().GetBoolFlag(
|
||||
platform::config_package_nearby::nearby_platform_feature::
|
||||
kEnableTaskScheduler)) {
|
||||
if (use_task_scheduler_) {
|
||||
absl::MutexLock lock(&mutex_);
|
||||
if ((delay < 0) || (interval < 0)) {
|
||||
NEARBY_LOGS(WARNING)
|
||||
@@ -95,9 +98,7 @@ bool Timer::Create(int delay, int interval,
|
||||
}
|
||||
|
||||
bool Timer::Stop() {
|
||||
if (NearbyFlags::GetInstance().GetBoolFlag(
|
||||
platform::config_package_nearby::nearby_platform_feature::
|
||||
kEnableTaskScheduler)) {
|
||||
if (use_task_scheduler_) {
|
||||
absl::MutexLock lock(&mutex_);
|
||||
if (cancelable_task_ == nullptr) {
|
||||
return true;
|
||||
|
||||
@@ -21,6 +21,7 @@
|
||||
#include <vector>
|
||||
|
||||
#include "absl/base/thread_annotations.h"
|
||||
#include "absl/functional/any_invocable.h"
|
||||
#include "absl/synchronization/mutex.h"
|
||||
#include "internal/platform/implementation/cancelable.h"
|
||||
#include "internal/platform/implementation/timer.h"
|
||||
@@ -32,7 +33,7 @@ namespace windows {
|
||||
|
||||
class Timer : public api::Timer {
|
||||
public:
|
||||
Timer() = default;
|
||||
Timer();
|
||||
~Timer() override;
|
||||
|
||||
bool Create(int delay, int interval,
|
||||
@@ -45,6 +46,7 @@ class Timer : public api::Timer {
|
||||
static void CALLBACK TimerRoutine(PVOID lpParam, BOOLEAN TimerOrWaitFired);
|
||||
|
||||
mutable absl::Mutex mutex_;
|
||||
const bool use_task_scheduler_;
|
||||
int delay_ ABSL_GUARDED_BY(mutex_);
|
||||
int interval_ ABSL_GUARDED_BY(mutex_);
|
||||
absl::AnyInvocable<void()> callback_;
|
||||
|
||||
@@ -15,15 +15,14 @@
|
||||
#include "internal/platform/implementation/timer.h"
|
||||
|
||||
#include <chrono> // NOLINT
|
||||
// NOLINT
|
||||
#include <memory>
|
||||
#include <thread> // NOLINT
|
||||
#include <vector>
|
||||
|
||||
#include "gtest/gtest.h"
|
||||
#include "absl/synchronization/notification.h"
|
||||
#include "absl/time/time.h"
|
||||
#include "internal/flags/nearby_flags.h"
|
||||
#include "internal/platform/count_down_latch.h"
|
||||
#include "internal/platform/flags/nearby_platform_feature_flags.h"
|
||||
#include "internal/platform/implementation/platform.h"
|
||||
|
||||
@@ -31,7 +30,7 @@ namespace nearby {
|
||||
namespace windows {
|
||||
namespace {
|
||||
|
||||
class TimerTaskSchedulerFlagTest : public ::testing::TestWithParam<bool> {
|
||||
class TimerTest : public ::testing::TestWithParam<bool> {
|
||||
public:
|
||||
void SetUp() override {
|
||||
NearbyFlags::GetInstance().OverrideBoolFlagValue(
|
||||
@@ -39,9 +38,13 @@ class TimerTaskSchedulerFlagTest : public ::testing::TestWithParam<bool> {
|
||||
kEnableTaskScheduler,
|
||||
GetParam());
|
||||
}
|
||||
|
||||
void TearDown() override {
|
||||
NearbyFlags::GetInstance().ResetOverridedValues();
|
||||
}
|
||||
};
|
||||
|
||||
TEST_P(TimerTaskSchedulerFlagTest, TestCreateTimer) {
|
||||
TEST_P(TimerTest, TestCreateTimer) {
|
||||
int count = 0;
|
||||
|
||||
std::unique_ptr<nearby::api::Timer> timer =
|
||||
@@ -53,20 +56,24 @@ TEST_P(TimerTaskSchedulerFlagTest, TestCreateTimer) {
|
||||
}
|
||||
|
||||
// This test case cannot run on Google3
|
||||
TEST_P(TimerTaskSchedulerFlagTest, TestRepeatTimer) {
|
||||
TEST_P(TimerTest, TestRepeatTimer) {
|
||||
CountDownLatch latch(3);
|
||||
int count = 0;
|
||||
|
||||
std::unique_ptr<nearby::api::Timer> timer =
|
||||
nearby::api::ImplementationPlatform::CreateTimer();
|
||||
|
||||
ASSERT_TRUE(timer != nullptr);
|
||||
EXPECT_TRUE(timer->Create(300, 300, [&]() { ++count; }));
|
||||
std::this_thread::sleep_for(std::chrono::seconds(1));
|
||||
EXPECT_TRUE(timer->Stop());
|
||||
EXPECT_TRUE(timer->Create(300, 300, [&]() {
|
||||
++count;
|
||||
latch.CountDown();
|
||||
}));
|
||||
|
||||
EXPECT_TRUE(latch.Await(absl::Seconds(2)));
|
||||
EXPECT_EQ(count, 3);
|
||||
EXPECT_TRUE(timer->Stop());
|
||||
}
|
||||
|
||||
TEST_P(TimerTaskSchedulerFlagTest, TestFireNow) {
|
||||
TEST_P(TimerTest, TestFireNow) {
|
||||
int count = 0;
|
||||
absl::Notification notification;
|
||||
|
||||
@@ -84,8 +91,8 @@ TEST_P(TimerTaskSchedulerFlagTest, TestFireNow) {
|
||||
EXPECT_EQ(count, 1);
|
||||
}
|
||||
|
||||
INSTANTIATE_TEST_SUITE_P(TimerTest, TimerTaskSchedulerFlagTest,
|
||||
testing::ValuesIn(std::vector<bool>{true, false}));
|
||||
INSTANTIATE_TEST_SUITE_P(TimerTaskSchedulerFlagTest, TimerTest,
|
||||
testing::Bool());
|
||||
|
||||
} // namespace
|
||||
} // namespace windows
|
||||
|
||||
@@ -31,6 +31,20 @@
|
||||
|
||||
namespace nearby {
|
||||
|
||||
class ScheduledExecutorTest : public ::testing::Test {
|
||||
public:
|
||||
void SetUp() override {
|
||||
NearbyFlags::GetInstance().OverrideBoolFlagValue(
|
||||
platform::config_package_nearby::nearby_platform_feature::
|
||||
kEnableTaskScheduler,
|
||||
true);
|
||||
}
|
||||
|
||||
void TearDown() override {
|
||||
NearbyFlags::GetInstance().ResetOverridedValues();
|
||||
}
|
||||
};
|
||||
|
||||
// kShortDelay must be significant enough to guarantee that OS under heavy load
|
||||
// should be able to execute the non-blocking test paths within this time.
|
||||
absl::Duration kShortDelay = absl::Milliseconds(100);
|
||||
@@ -39,17 +53,11 @@ absl::Duration kShortDelay = absl::Milliseconds(100);
|
||||
// will let kShortDelay fire and jobs scheduled before the kLongDelay fires.
|
||||
absl::Duration kLongDelay = 10 * kShortDelay;
|
||||
|
||||
TEST(ScheduledExecutorTest, ConsructorDestructorWorks) {
|
||||
NearbyFlags::GetInstance().OverrideBoolFlagValue(
|
||||
platform::config_package_nearby::nearby_platform_feature::
|
||||
kEnableTaskScheduler, true);
|
||||
TEST_F(ScheduledExecutorTest, ConsructorDestructorWorks) {
|
||||
ScheduledExecutor executor;
|
||||
}
|
||||
|
||||
TEST(ScheduledExecutorTest, CanExecute) {
|
||||
NearbyFlags::GetInstance().OverrideBoolFlagValue(
|
||||
platform::config_package_nearby::nearby_platform_feature::
|
||||
kEnableTaskScheduler, true);
|
||||
TEST_F(ScheduledExecutorTest, CanExecute) {
|
||||
absl::Mutex mutex;
|
||||
absl::CondVar cond;
|
||||
std::atomic_bool done = false;
|
||||
@@ -67,10 +75,7 @@ TEST(ScheduledExecutorTest, CanExecute) {
|
||||
EXPECT_TRUE(done);
|
||||
}
|
||||
|
||||
TEST(ScheduledExecutorTest, CanSchedule) {
|
||||
NearbyFlags::GetInstance().OverrideBoolFlagValue(
|
||||
platform::config_package_nearby::nearby_platform_feature::
|
||||
kEnableTaskScheduler, true);
|
||||
TEST_F(ScheduledExecutorTest, CanSchedule) {
|
||||
ScheduledExecutor executor;
|
||||
std::atomic_int value = 0;
|
||||
absl::Mutex mutex;
|
||||
@@ -98,10 +103,7 @@ TEST(ScheduledExecutorTest, CanSchedule) {
|
||||
EXPECT_EQ(value, 5);
|
||||
}
|
||||
|
||||
TEST(ScheduledExecutorTest, CanCancel) {
|
||||
NearbyFlags::GetInstance().OverrideBoolFlagValue(
|
||||
platform::config_package_nearby::nearby_platform_feature::
|
||||
kEnableTaskScheduler, true);
|
||||
TEST_F(ScheduledExecutorTest, CanCancel) {
|
||||
ScheduledExecutor executor;
|
||||
std::atomic_int value = 0;
|
||||
Cancelable cancelable =
|
||||
@@ -112,10 +114,7 @@ TEST(ScheduledExecutorTest, CanCancel) {
|
||||
EXPECT_EQ(value, 0);
|
||||
}
|
||||
|
||||
TEST(ScheduledExecutorTest, CanCancelTwice) {
|
||||
NearbyFlags::GetInstance().OverrideBoolFlagValue(
|
||||
platform::config_package_nearby::nearby_platform_feature::
|
||||
kEnableTaskScheduler, true);
|
||||
TEST_F(ScheduledExecutorTest, CanCancelTwice) {
|
||||
ScheduledExecutor executor;
|
||||
std::atomic_int value = 0;
|
||||
Cancelable cancelable =
|
||||
@@ -129,10 +128,7 @@ TEST(ScheduledExecutorTest, CanCancelTwice) {
|
||||
EXPECT_EQ(value, 0);
|
||||
}
|
||||
|
||||
TEST(ScheduledExecutorTest, FailToCancel) {
|
||||
NearbyFlags::GetInstance().OverrideBoolFlagValue(
|
||||
platform::config_package_nearby::nearby_platform_feature::
|
||||
kEnableTaskScheduler, true);
|
||||
TEST_F(ScheduledExecutorTest, FailToCancel) {
|
||||
absl::Mutex mutex;
|
||||
absl::CondVar cond;
|
||||
ScheduledExecutor executor;
|
||||
@@ -155,11 +151,8 @@ TEST(ScheduledExecutorTest, FailToCancel) {
|
||||
EXPECT_EQ(value, 1);
|
||||
}
|
||||
|
||||
TEST(ScheduledExecutorTest,
|
||||
CancelWhileRunning_TaskCompletesBeforeCancelReturns) {
|
||||
NearbyFlags::GetInstance().OverrideBoolFlagValue(
|
||||
platform::config_package_nearby::nearby_platform_feature::
|
||||
kEnableTaskScheduler, true);
|
||||
TEST_F(ScheduledExecutorTest,
|
||||
CancelWhileRunning_TaskCompletesBeforeCancelReturns) {
|
||||
CountDownLatch start_latch(1);
|
||||
ScheduledExecutor executor;
|
||||
std::atomic_int value = 0;
|
||||
@@ -178,11 +171,8 @@ TEST(ScheduledExecutorTest,
|
||||
EXPECT_EQ(value, 1);
|
||||
}
|
||||
|
||||
TEST(ScheduledExecutorTest,
|
||||
CancelTwiceWhileRunning_TaskCompletesBeforeCancelReturns) {
|
||||
NearbyFlags::GetInstance().OverrideBoolFlagValue(
|
||||
platform::config_package_nearby::nearby_platform_feature::
|
||||
kEnableTaskScheduler, true);
|
||||
TEST_F(ScheduledExecutorTest,
|
||||
CancelTwiceWhileRunning_TaskCompletesBeforeCancelReturns) {
|
||||
CountDownLatch start_latch(1);
|
||||
ScheduledExecutor executor;
|
||||
std::atomic_int value = 0;
|
||||
@@ -203,10 +193,7 @@ TEST(ScheduledExecutorTest,
|
||||
EXPECT_EQ(value, 1);
|
||||
}
|
||||
|
||||
TEST(ScheduledExecutorTest, ShutdownWaitsForRunningTasks) {
|
||||
NearbyFlags::GetInstance().OverrideBoolFlagValue(
|
||||
platform::config_package_nearby::nearby_platform_feature::
|
||||
kEnableTaskScheduler, true);
|
||||
TEST_F(ScheduledExecutorTest, ShutdownWaitsForRunningTasks) {
|
||||
ScheduledExecutor executor;
|
||||
std::atomic_int value = 0;
|
||||
executor.Execute([&]() {
|
||||
@@ -219,20 +206,14 @@ TEST(ScheduledExecutorTest, ShutdownWaitsForRunningTasks) {
|
||||
EXPECT_EQ(value, 1);
|
||||
}
|
||||
|
||||
TEST(ScheduledExecutorTest, ExecuteAfterShutdownFails) {
|
||||
NearbyFlags::GetInstance().OverrideBoolFlagValue(
|
||||
platform::config_package_nearby::nearby_platform_feature::
|
||||
kEnableTaskScheduler, true);
|
||||
TEST_F(ScheduledExecutorTest, ExecuteAfterShutdownFails) {
|
||||
ScheduledExecutor executor;
|
||||
|
||||
executor.Shutdown();
|
||||
executor.Execute([&]() { FAIL() << "Task should not run"; });
|
||||
}
|
||||
|
||||
TEST(ScheduledExecutorTest, ExecuteDuringShutdownFails) {
|
||||
NearbyFlags::GetInstance().OverrideBoolFlagValue(
|
||||
platform::config_package_nearby::nearby_platform_feature::
|
||||
kEnableTaskScheduler, true);
|
||||
TEST_F(ScheduledExecutorTest, ExecuteDuringShutdownFails) {
|
||||
CountDownLatch latch(1);
|
||||
ScheduledExecutor executor;
|
||||
|
||||
@@ -245,10 +226,7 @@ TEST(ScheduledExecutorTest, ExecuteDuringShutdownFails) {
|
||||
executor.Shutdown();
|
||||
}
|
||||
|
||||
TEST(ScheduledExecutorTest, SimulatedClockCanSchedule) {
|
||||
NearbyFlags::GetInstance().OverrideBoolFlagValue(
|
||||
platform::config_package_nearby::nearby_platform_feature::
|
||||
kEnableTaskScheduler, true);
|
||||
TEST_F(ScheduledExecutorTest, SimulatedClockCanSchedule) {
|
||||
MediumEnvironment::Instance().Start({.use_simulated_clock = true});
|
||||
FakeClock* fake_clock =
|
||||
MediumEnvironment::Instance().GetSimulatedClock().value();
|
||||
@@ -286,11 +264,8 @@ TEST(ScheduledExecutorTest, SimulatedClockCanSchedule) {
|
||||
MediumEnvironment::Instance().Stop();
|
||||
}
|
||||
|
||||
TEST(ScheduledExecutorTest,
|
||||
DestroyExecutorWithSimulatedClockIgnoresPendingTasks) {
|
||||
NearbyFlags::GetInstance().OverrideBoolFlagValue(
|
||||
platform::config_package_nearby::nearby_platform_feature::
|
||||
kEnableTaskScheduler, true);
|
||||
TEST_F(ScheduledExecutorTest,
|
||||
DestroyExecutorWithSimulatedClockIgnoresPendingTasks) {
|
||||
MediumEnvironment::Instance().Start({.use_simulated_clock = true});
|
||||
FakeClock* fake_clock =
|
||||
MediumEnvironment::Instance().GetSimulatedClock().value();
|
||||
@@ -307,7 +282,7 @@ TEST(ScheduledExecutorTest,
|
||||
MediumEnvironment::Instance().Stop();
|
||||
}
|
||||
|
||||
struct ThreadCheckTestClass {
|
||||
struct ScheduledThreadCheckTestClass {
|
||||
ScheduledExecutor executor;
|
||||
int value ABSL_GUARDED_BY(executor) = 0;
|
||||
|
||||
@@ -315,35 +290,29 @@ struct ThreadCheckTestClass {
|
||||
int getValue() ABSL_EXCLUSIVE_LOCKS_REQUIRED(executor) { return value; }
|
||||
};
|
||||
|
||||
TEST(ScheduledExecutorTest, ThreadCheck_Execute) {
|
||||
NearbyFlags::GetInstance().OverrideBoolFlagValue(
|
||||
platform::config_package_nearby::nearby_platform_feature::
|
||||
kEnableTaskScheduler, true);
|
||||
ThreadCheckTestClass test_class;
|
||||
TEST_F(ScheduledExecutorTest, ThreadCheck_Execute) {
|
||||
ScheduledThreadCheckTestClass test_class;
|
||||
absl::Notification notification;
|
||||
|
||||
test_class.executor.Execute(
|
||||
[&test_class, ¬ification]() ABSL_EXCLUSIVE_LOCKS_REQUIRED(
|
||||
test_class.executor) {
|
||||
test_class.incValue();
|
||||
notification.Notify();
|
||||
});
|
||||
[&test_class, ¬ification]()
|
||||
ABSL_EXCLUSIVE_LOCKS_REQUIRED(test_class.executor) {
|
||||
test_class.incValue();
|
||||
notification.Notify();
|
||||
});
|
||||
EXPECT_TRUE(notification.WaitForNotificationWithTimeout(absl::Seconds(2)));
|
||||
}
|
||||
|
||||
TEST(ScheduledExecutorTest, ThreadCheck_Schedule) {
|
||||
NearbyFlags::GetInstance().OverrideBoolFlagValue(
|
||||
platform::config_package_nearby::nearby_platform_feature::
|
||||
kEnableTaskScheduler, true);
|
||||
ThreadCheckTestClass test_class;
|
||||
TEST_F(ScheduledExecutorTest, ThreadCheck_Schedule) {
|
||||
ScheduledThreadCheckTestClass test_class;
|
||||
absl::Notification notification;
|
||||
|
||||
test_class.executor.Schedule(
|
||||
[&test_class, ¬ification]() ABSL_EXCLUSIVE_LOCKS_REQUIRED(
|
||||
test_class.executor) {
|
||||
test_class.incValue();
|
||||
notification.Notify();
|
||||
},
|
||||
[&test_class, ¬ification]()
|
||||
ABSL_EXCLUSIVE_LOCKS_REQUIRED(test_class.executor) {
|
||||
test_class.incValue();
|
||||
notification.Notify();
|
||||
},
|
||||
absl::ZeroDuration());
|
||||
EXPECT_TRUE(notification.WaitForNotificationWithTimeout(absl::Seconds(2)));
|
||||
}
|
||||
|
||||
@@ -19,8 +19,16 @@
|
||||
#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"
|
||||
@@ -36,10 +44,20 @@ class SettableFuture : public api::SettableFuture<T> {
|
||||
|
||||
// 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}); });
|
||||
: timer_(std::make_unique<TimerImpl>()) {
|
||||
timer_->Start(absl::ToInt64Milliseconds(timeout), 0, [this] {
|
||||
// Offload the timeout to a single thread executor.
|
||||
if (NearbyFlags::GetInstance().GetBoolFlag(
|
||||
platform::config_package_nearby::nearby_platform_feature::
|
||||
kEnableTaskScheduler)) {
|
||||
executor_ = api::ImplementationPlatform::CreateSingleThreadExecutor();
|
||||
executor_->Execute([this]() { SetException({Exception::kTimeout}); });
|
||||
} else {
|
||||
SetException({Exception::kTimeout});
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
~SettableFuture() override = default;
|
||||
|
||||
bool Set(T value) override {
|
||||
@@ -148,6 +166,7 @@ class SettableFuture : public api::SettableFuture<T> {
|
||||
T value_;
|
||||
Exception exception_{Exception::kFailed};
|
||||
std::unique_ptr<TimerImpl> timer_;
|
||||
std::unique_ptr<api::SubmittableExecutor> executor_;
|
||||
};
|
||||
|
||||
} // namespace nearby
|
||||
|
||||
@@ -16,13 +16,16 @@
|
||||
|
||||
#include <atomic>
|
||||
#include <string>
|
||||
#include <vector>
|
||||
|
||||
#include "gtest/gtest.h"
|
||||
#include "absl/base/thread_annotations.h"
|
||||
#include "absl/synchronization/mutex.h"
|
||||
#include "absl/time/clock.h"
|
||||
#include "absl/time/time.h"
|
||||
#include "internal/platform/count_down_latch.h"
|
||||
#include "internal/platform/exception.h"
|
||||
#include "internal/platform/future.h"
|
||||
|
||||
namespace nearby {
|
||||
|
||||
|
||||
Reference in New Issue
Block a user