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

Cleanup enable_task_scheduler flag.

Browse Source 
PiperOrigin-RevId: 738938572
This commit is contained in:
Francis Tsui
2025-03-20 14:21:05 -07:00
committed by Copybara-Service
parent a3049fe436
commit 1750168761
9 changed files with 63 additions and 298 deletions
Download Patch File Download Diff File Expand all files Collapse all files
internal/platform/flags/nearby_platform_feature_flags.h
-4
View File
@@ -73,10 +73,6 @@ constexpr auto kEnableIntelPieSdk =
constexpr auto kEnableNewBluetoothRefactor =
flags::Flag<bool>(kConfigPackage, "45615156", false);
// Enable/Disable task scheduler for ScheduledExecutor and timer
constexpr auto kEnableTaskScheduler =
flags::Flag<bool>(kConfigPackage, "45643835", false);
// Enable/Disable Wi-Fi hotspot native
constexpr auto kEnableWifiHotspotNative =
flags::Flag<bool>(kConfigPackage, "45667396", false);
internal/platform/implementation/windows/scheduled_executor.cc
+11 -55
View File
@@ -20,8 +20,6 @@
#include <utility>
#include "absl/time/time.h"
#include "internal/flags/nearby_flags.h"
#include "internal/platform/flags/nearby_platform_feature_flags.h"
#include "internal/platform/implementation/cancelable.h"
#include "internal/platform/logging.h"
#include "internal/platform/runnable.h"
@@ -31,10 +29,7 @@ namespace windows {
ScheduledExecutor::ScheduledExecutor()
: executor_(std::make_unique<nearby::windows::Executor>()),
shut_down_(false),
use_task_scheduler_(NearbyFlags::GetInstance().GetBoolFlag(
platform::config_package_nearby::nearby_platform_feature::
kEnableTaskScheduler)) {}
shut_down_(false) {}
// 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.
@@ -42,39 +37,13 @@ ScheduledExecutor::ScheduledExecutor()
// using std:shared_ptr<> instead of std::unique_ptr<>.
std::shared_ptr<api::Cancelable> ScheduledExecutor::Schedule(
Runnable&& runnable, absl::Duration duration) {
if (use_task_scheduler_) {
if (shut_down_) {
LOG(ERROR) << __func__
<< ": Attempt to Schedule on a shut down executor.";
if (shut_down_) {
LOG(ERROR) << __func__
<< ": Attempt to Schedule on a shut down executor.";
return nullptr;
}
return task_scheduler_.Schedule(std::move(runnable), duration);
} else {
if (shut_down_) {
LOG(ERROR) << __func__
<< ": Attempt to Schedule on a shut down executor.";
return nullptr;
}
// Cleans completed tasks
auto it = scheduled_tasks_.begin();
while (it != scheduled_tasks_.end()) {
if ((*it)->IsDone()) {
it = scheduled_tasks_.erase(it);
} else {
++it;
}
}
std::shared_ptr<ScheduledTask> task =
std::make_shared<ScheduledTask>(std::move(runnable), duration);
scheduled_tasks_.push_back(task);
executor_->Execute([task]() { task->Start(); });
return task;
return nullptr;
}
return task_scheduler_.Schedule(std::move(runnable), duration);
}
void ScheduledExecutor::Execute(Runnable&& runnable) {
@@ -87,24 +56,11 @@ void ScheduledExecutor::Execute(Runnable&& runnable) {
}
void ScheduledExecutor::Shutdown() {
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;
}
if (!shut_down_) {
shut_down_ = true;
executor_->Shutdown();
task_scheduler_.Shutdown();
return;
}
LOG(ERROR) << __func__ << ": Attempt to Shutdown on a shut down executor.";
}
internal/platform/implementation/windows/scheduled_executor.h
-41
View File
@@ -19,9 +19,7 @@
#include <atomic>
#include <memory>
#include <vector>
#include "absl/synchronization/notification.h"
#include "absl/time/time.h"
#include "internal/platform/implementation/cancelable.h"
#include "internal/platform/implementation/scheduled_executor.h"
@@ -32,8 +30,6 @@
namespace nearby {
namespace windows {
#define TIMER_NAME_BUFFER_SIZE 64
// An Executor that can schedule commands to run after a given delay, or to
// execute periodically.
//
@@ -58,46 +54,9 @@ class ScheduledExecutor : public api::ScheduledExecutor {
void Shutdown() override;
private:
class ScheduledTask : public api::Cancelable {
public:
explicit ScheduledTask(Runnable&& task, absl::Duration duration)
: task_(std::move(task)), duration_(duration) {}
bool Cancel() override {
if (is_executed_ || is_cancelled_) {
return false;
}
is_cancelled_ = true;
notification_.Notify();
return true;
};
void Start() {
if (is_executed_ ||
notification_.WaitForNotificationWithTimeout(duration_)) {
return;
}
is_executed_ = true;
task_();
}
bool IsDone() const { return is_cancelled_ || is_executed_; }
private:
Runnable task_;
absl::Duration duration_;
absl::Notification notification_;
bool is_cancelled_ = false;
bool is_executed_ = false;
};
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_;
};
internal/platform/implementation/windows/scheduled_executor_test.cc
+4 -23
View File
@@ -22,8 +22,6 @@
#include "absl/synchronization/notification.h"
#include "absl/time/clock.h"
#include "absl/time/time.h"
#include "internal/flags/nearby_flags.h"
#include "internal/platform/flags/nearby_platform_feature_flags.h"
#include "internal/platform/implementation/windows/test_data.h"
namespace nearby {
@@ -32,21 +30,7 @@ namespace {
constexpr absl::Duration kWaitTimeout = absl::Milliseconds(2000);
class ScheduledExecutorTest : public ::testing::TestWithParam<bool> {
public:
void SetUp() override {
NearbyFlags::GetInstance().OverrideBoolFlagValue(
platform::config_package_nearby::nearby_platform_feature::
kEnableTaskScheduler,
GetParam());
}
void TearDown() override {
NearbyFlags::GetInstance().ResetOverridedValues();
}
};
TEST_P(ScheduledExecutorTest, ExecuteSucceeds) {
TEST(ScheduledExecutorTest, ExecuteSucceeds) {
absl::Notification notification;
// Arrange
std::string expected(RUNNABLE_0_TEXT.c_str());
@@ -79,7 +63,7 @@ TEST_P(ScheduledExecutorTest, ExecuteSucceeds) {
ASSERT_EQ(output, expected);
}
TEST_P(ScheduledExecutorTest, ScheduleSucceeds) {
TEST(ScheduledExecutorTest, ScheduleSucceeds) {
absl::Notification notification;
// Arrange
std::string expected(RUNNABLE_0_TEXT.c_str());
@@ -114,7 +98,7 @@ TEST_P(ScheduledExecutorTest, ScheduleSucceeds) {
ASSERT_EQ(output, expected);
}
TEST_P(ScheduledExecutorTest, CancelSucceeds) {
TEST(ScheduledExecutorTest, CancelSucceeds) {
absl::Notification notification;
// Arrange
std::string expected("");
@@ -150,7 +134,7 @@ TEST_P(ScheduledExecutorTest, CancelSucceeds) {
ASSERT_EQ(output, expected);
}
TEST_P(ScheduledExecutorTest, CancelAfterStartedFails) {
TEST(ScheduledExecutorTest, CancelAfterStartedFails) {
absl::Notification notification;
// Arrange
std::string expected(RUNNABLE_0_TEXT.c_str());
@@ -187,9 +171,6 @@ TEST_P(ScheduledExecutorTest, CancelAfterStartedFails) {
ASSERT_EQ(output, expected);
}
INSTANTIATE_TEST_SUITE_P(ScheduledExecutorTaskSchedulerFlagTest,
ScheduledExecutorTest, testing::Bool());
} // namespace
} // namespace windows
} // namespace nearby
internal/platform/implementation/windows/timer.cc
+24 -100
View File
@@ -21,114 +21,46 @@
#include "absl/functional/any_invocable.h"
#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"
#include "internal/platform/logging.h"
#include "internal/platform/runnable.h"
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 (use_task_scheduler_) {
absl::MutexLock lock(&mutex_);
if ((delay < 0) || (interval < 0)) {
LOG(WARNING) << "Delay and interval shouldn\'t be negative value.";
return false;
}
if (cancelable_task_) {
return false;
}
callback_ = std::move(callback);
std::function<void()> internal_callback = [this]() {
if (callback_ != nullptr) {
callback_();
}
};
cancelable_task_ = task_scheduler_.Schedule(std::move(internal_callback),
absl::Milliseconds(delay),
absl::Milliseconds(interval));
return cancelable_task_ != nullptr;
} else {
absl::MutexLock lock(&mutex_);
if ((delay < 0) || (interval < 0)) {
LOG(WARNING) << "Delay and interval shouldn\'t be negative value.";
return false;
}
if (timer_queue_handle_ != nullptr) {
return false;
}
timer_queue_handle_ = CreateTimerQueue();
if (timer_queue_handle_ == nullptr) {
LOG(ERROR) << "Failed to create timer queue.";
return false;
}
delay_ = delay;
interval_ = interval;
callback_ = std::move(callback);
if (!CreateTimerQueueTimer(&handle_, timer_queue_handle_,
static_cast<WAITORTIMERCALLBACK>(TimerRoutine),
&callback_, delay, interval,
WT_EXECUTEDEFAULT)) {
if (!DeleteTimerQueueEx(timer_queue_handle_, nullptr)) {
LOG(ERROR) << "Failed to create timer in timer queue.";
}
timer_queue_handle_ = nullptr;
return false;
}
return true;
absl::MutexLock lock(&mutex_);
if ((delay < 0) || (interval < 0)) {
LOG(WARNING) << "Delay and interval shouldn\'t be negative value.";
return false;
}
if (cancelable_task_) {
return false;
}
callback_ = std::move(callback);
std::function<void()> internal_callback = [this]() {
if (callback_ != nullptr) {
callback_();
}
};
cancelable_task_ = task_scheduler_.Schedule(std::move(internal_callback),
absl::Milliseconds(delay),
absl::Milliseconds(interval));
return cancelable_task_ != nullptr;
}
bool Timer::Stop() {
if (use_task_scheduler_) {
absl::MutexLock lock(&mutex_);
if (cancelable_task_ == nullptr) {
return true;
}
bool result = cancelable_task_->Cancel();
cancelable_task_ = nullptr;
return result;
} else {
absl::MutexLock lock(&mutex_);
if (timer_queue_handle_ == nullptr) {
return true;
}
if (!DeleteTimerQueueTimer(timer_queue_handle_, handle_, nullptr)) {
if (GetLastError() != ERROR_IO_PENDING) {
LOG(ERROR) << "Failed to delete timer from timer queue.";
return false;
}
}
handle_ = nullptr;
if (!DeleteTimerQueueEx(timer_queue_handle_, nullptr)) {
LOG(ERROR) << "Failed to delete timer queue.";
return false;
}
timer_queue_handle_ = nullptr;
absl::MutexLock lock(&mutex_);
if (cancelable_task_ == nullptr) {
return true;
}
bool result = cancelable_task_->Cancel();
cancelable_task_ = nullptr;
return result;
}
bool Timer::FireNow() {
@@ -153,13 +85,5 @@ bool Timer::FireNow() {
return true;
}
void CALLBACK Timer::TimerRoutine(PVOID lpParam, BOOLEAN TimerOrWaitFired) {
absl::AnyInvocable<void()>* callback =
reinterpret_cast<absl::AnyInvocable<void()>*>(lpParam);
if (*callback != nullptr) {
(*callback)();
}
}
} // namespace windows
} // namespace nearby
internal/platform/implementation/windows/timer.h
+1 -9
View File
@@ -18,7 +18,6 @@
#include <windows.h>
#include <memory>
#include <vector>
#include "absl/base/thread_annotations.h"
#include "absl/functional/any_invocable.h"
@@ -33,7 +32,7 @@ namespace windows {
class Timer : public api::Timer {
public:
Timer();
Timer() = default;
~Timer() override;
bool Create(int delay, int interval,
@@ -43,15 +42,8 @@ class Timer : public api::Timer {
bool FireNow() override ABSL_LOCKS_EXCLUDED(mutex_);
private:
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_;
HANDLE handle_ ABSL_GUARDED_BY(mutex_) = nullptr;
HANDLE timer_queue_handle_ ABSL_GUARDED_BY(mutex_) = nullptr;
std::unique_ptr<SubmittableExecutor> task_executor_ ABSL_GUARDED_BY(mutex_) =
nullptr;
TaskScheduler task_scheduler_ ABSL_GUARDED_BY(mutex_);
internal/platform/implementation/windows/timer_test.cc
+3 -24
View File
@@ -14,37 +14,19 @@
#include "internal/platform/implementation/timer.h"
#include <chrono> // NOLINT
#include <memory>
#include <thread> // NOLINT
#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"
namespace nearby {
namespace windows {
namespace {
class TimerTest : public ::testing::TestWithParam<bool> {
public:
void SetUp() override {
NearbyFlags::GetInstance().OverrideBoolFlagValue(
platform::config_package_nearby::nearby_platform_feature::
kEnableTaskScheduler,
GetParam());
}
void TearDown() override {
NearbyFlags::GetInstance().ResetOverridedValues();
}
};
TEST_P(TimerTest, TestCreateTimer) {
TEST(TimerTest, TestCreateTimer) {
int count = 0;
std::unique_ptr<nearby::api::Timer> timer =
@@ -56,7 +38,7 @@ TEST_P(TimerTest, TestCreateTimer) {
}
// This test case cannot run on Google3
TEST_P(TimerTest, TestRepeatTimer) {
TEST(TimerTest, TestRepeatTimer) {
CountDownLatch latch(3);
int count = 0;
std::unique_ptr<nearby::api::Timer> timer =
@@ -73,7 +55,7 @@ TEST_P(TimerTest, TestRepeatTimer) {
EXPECT_TRUE(timer->Stop());
}
TEST_P(TimerTest, TestFireNow) {
TEST(TimerTest, TestFireNow) {
int count = 0;
absl::Notification notification;
@@ -91,9 +73,6 @@ TEST_P(TimerTest, TestFireNow) {
EXPECT_EQ(count, 1);
}
INSTANTIATE_TEST_SUITE_P(TimerTaskSchedulerFlagTest, TimerTest,
testing::Bool());
} // namespace
} // namespace windows
} // namespace nearby
internal/platform/scheduled_executor_test.cc
+18 -34
View File
@@ -22,29 +22,13 @@
#include "absl/synchronization/notification.h"
#include "absl/time/clock.h"
#include "absl/time/time.h"
#include "internal/flags/nearby_flags.h"
#include "internal/platform/cancelable.h"
#include "internal/platform/count_down_latch.h"
#include "internal/platform/flags/nearby_platform_feature_flags.h"
#include "internal/platform/medium_environment.h"
#include "internal/test/fake_clock.h"
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);
@@ -53,11 +37,11 @@ absl::Duration kShortDelay = absl::Milliseconds(100);
// will let kShortDelay fire and jobs scheduled before the kLongDelay fires.
absl::Duration kLongDelay = 10 * kShortDelay;
TEST_F(ScheduledExecutorTest, ConsructorDestructorWorks) {
TEST(ScheduledExecutorTest, ConsructorDestructorWorks) {
ScheduledExecutor executor;
}
TEST_F(ScheduledExecutorTest, CanExecute) {
TEST(ScheduledExecutorTest, CanExecute) {
absl::Mutex mutex;
absl::CondVar cond;
std::atomic_bool done = false;
@@ -75,7 +59,7 @@ TEST_F(ScheduledExecutorTest, CanExecute) {
EXPECT_TRUE(done);
}
TEST_F(ScheduledExecutorTest, CanSchedule) {
TEST(ScheduledExecutorTest, CanSchedule) {
ScheduledExecutor executor;
std::atomic_int value = 0;
absl::Mutex mutex;
@@ -103,7 +87,7 @@ TEST_F(ScheduledExecutorTest, CanSchedule) {
EXPECT_EQ(value, 5);
}
TEST_F(ScheduledExecutorTest, CanCancel) {
TEST(ScheduledExecutorTest, CanCancel) {
ScheduledExecutor executor;
std::atomic_int value = 0;
Cancelable cancelable =
@@ -114,7 +98,7 @@ TEST_F(ScheduledExecutorTest, CanCancel) {
EXPECT_EQ(value, 0);
}
TEST_F(ScheduledExecutorTest, CanCancelTwice) {
TEST(ScheduledExecutorTest, CanCancelTwice) {
ScheduledExecutor executor;
std::atomic_int value = 0;
Cancelable cancelable =
@@ -128,7 +112,7 @@ TEST_F(ScheduledExecutorTest, CanCancelTwice) {
EXPECT_EQ(value, 0);
}
TEST_F(ScheduledExecutorTest, FailToCancel) {
TEST(ScheduledExecutorTest, FailToCancel) {
absl::Mutex mutex;
absl::CondVar cond;
ScheduledExecutor executor;
@@ -151,8 +135,8 @@ TEST_F(ScheduledExecutorTest, FailToCancel) {
EXPECT_EQ(value, 1);
}
TEST_F(ScheduledExecutorTest,
CancelWhileRunning_TaskCompletesBeforeCancelReturns) {
TEST(ScheduledExecutorTest,
CancelWhileRunning_TaskCompletesBeforeCancelReturns) {
CountDownLatch start_latch(1);
ScheduledExecutor executor;
std::atomic_int value = 0;
@@ -171,8 +155,8 @@ TEST_F(ScheduledExecutorTest,
EXPECT_EQ(value, 1);
}
TEST_F(ScheduledExecutorTest,
CancelTwiceWhileRunning_TaskCompletesBeforeCancelReturns) {
TEST(ScheduledExecutorTest,
CancelTwiceWhileRunning_TaskCompletesBeforeCancelReturns) {
CountDownLatch start_latch(1);
ScheduledExecutor executor;
std::atomic_int value = 0;
@@ -193,7 +177,7 @@ TEST_F(ScheduledExecutorTest,
EXPECT_EQ(value, 1);
}
TEST_F(ScheduledExecutorTest, ShutdownWaitsForRunningTasks) {
TEST(ScheduledExecutorTest, ShutdownWaitsForRunningTasks) {
ScheduledExecutor executor;
std::atomic_int value = 0;
executor.Execute([&]() {
@@ -206,14 +190,14 @@ TEST_F(ScheduledExecutorTest, ShutdownWaitsForRunningTasks) {
EXPECT_EQ(value, 1);
}
TEST_F(ScheduledExecutorTest, ExecuteAfterShutdownFails) {
TEST(ScheduledExecutorTest, ExecuteAfterShutdownFails) {
ScheduledExecutor executor;
executor.Shutdown();
executor.Execute([&]() { FAIL() << "Task should not run"; });
}
TEST_F(ScheduledExecutorTest, ExecuteDuringShutdownFails) {
TEST(ScheduledExecutorTest, ExecuteDuringShutdownFails) {
CountDownLatch latch(1);
ScheduledExecutor executor;
@@ -226,7 +210,7 @@ TEST_F(ScheduledExecutorTest, ExecuteDuringShutdownFails) {
executor.Shutdown();
}
TEST_F(ScheduledExecutorTest, SimulatedClockCanSchedule) {
TEST(ScheduledExecutorTest, SimulatedClockCanSchedule) {
MediumEnvironment::Instance().Start({.use_simulated_clock = true});
FakeClock* fake_clock =
MediumEnvironment::Instance().GetSimulatedClock().value();
@@ -264,8 +248,8 @@ TEST_F(ScheduledExecutorTest, SimulatedClockCanSchedule) {
MediumEnvironment::Instance().Stop();
}
TEST_F(ScheduledExecutorTest,
DestroyExecutorWithSimulatedClockIgnoresPendingTasks) {
TEST(ScheduledExecutorTest,
DestroyExecutorWithSimulatedClockIgnoresPendingTasks) {
MediumEnvironment::Instance().Start({.use_simulated_clock = true});
FakeClock* fake_clock =
MediumEnvironment::Instance().GetSimulatedClock().value();
@@ -290,7 +274,7 @@ struct ScheduledThreadCheckTestClass {
int getValue() ABSL_EXCLUSIVE_LOCKS_REQUIRED(executor) { return value; }
};
TEST_F(ScheduledExecutorTest, ThreadCheck_Execute) {
TEST(ScheduledExecutorTest, ThreadCheck_Execute) {
ScheduledThreadCheckTestClass test_class;
absl::Notification notification;
@@ -303,7 +287,7 @@ TEST_F(ScheduledExecutorTest, ThreadCheck_Execute) {
EXPECT_TRUE(notification.WaitForNotificationWithTimeout(absl::Seconds(2)));
}
TEST_F(ScheduledExecutorTest, ThreadCheck_Schedule) {
TEST(ScheduledExecutorTest, ThreadCheck_Schedule) {
ScheduledThreadCheckTestClass test_class;
absl::Notification notification;
internal/platform/settable_future.h
+2 -8
View File
@@ -47,14 +47,8 @@ class SettableFuture : public api::SettableFuture<T> {
: 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});
}
executor_ = api::ImplementationPlatform::CreateSingleThreadExecutor();
executor_->Execute([this]() { SetException({Exception::kTimeout}); });
});
}