From d30f53f92bcc76da002fb592abfcc69a2962d584 Mon Sep 17 00:00:00 2001 From: Guogang Li Date: Tue, 16 Jul 2024 09:18:53 -0700 Subject: [PATCH] Added a class to schedule multiple tasks PiperOrigin-RevId: 652871095 --- .../platform/implementation/windows/BUILD | 3 + .../implementation/windows/task_scheduler.cc | 148 ++++++++++++++++++ .../implementation/windows/task_scheduler.h | 78 +++++++++ .../windows/task_scheduler_test.cc | 113 +++++++++++++ 4 files changed, 342 insertions(+) create mode 100644 internal/platform/implementation/windows/task_scheduler.cc create mode 100644 internal/platform/implementation/windows/task_scheduler.h create mode 100644 internal/platform/implementation/windows/task_scheduler_test.cc diff --git a/internal/platform/implementation/windows/BUILD b/internal/platform/implementation/windows/BUILD index 9b38cf0a..1c3ada7b 100644 --- a/internal/platform/implementation/windows/BUILD +++ b/internal/platform/implementation/windows/BUILD @@ -38,6 +38,7 @@ cc_library( "scheduled_executor.h", "settable_future.h", "submittable_executor.h", + "task_scheduler.h", "timer.h", "utils.h", ], @@ -179,6 +180,7 @@ cc_library( "session_manager.cc", "submittable_executor.cc", "system_clock.cc", + "task_scheduler.cc", "thread_pool.cc", "utils.cc", "webrtc.cc", @@ -287,6 +289,7 @@ cc_test( "preferences_repository_test.cc", "scheduled_executor_test.cc", "submittable_executor_test.cc", + "task_scheduler_test.cc", "thread_pool_test.cc", "timer_test.cc", "utils_test.cc", diff --git a/internal/platform/implementation/windows/task_scheduler.cc b/internal/platform/implementation/windows/task_scheduler.cc new file mode 100644 index 00000000..d1edf156 --- /dev/null +++ b/internal/platform/implementation/windows/task_scheduler.cc @@ -0,0 +1,148 @@ +// Copyright 2024 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. + +#include "internal/platform/implementation/windows/task_scheduler.h" + +#include + +#include +#include +#include + +#include "absl/time/time.h" +#include "internal/platform/implementation/cancelable.h" +#include "internal/platform/logging.h" +#include "internal/platform/mutex_lock.h" +#include "internal/platform/runnable.h" + +namespace nearby::windows { +namespace { +void CALLBACK TimerRoutine(PVOID lpParam, BOOLEAN TimerOrWaitFired) { + Runnable* task = reinterpret_cast(lpParam); + if (task != nullptr) { + (*task)(); + } +} +} // namespace + +TaskScheduler::TaskScheduler() { + NEARBY_LOGS(INFO) << __func__ << ": Created task scheduler: " << this; +} +TaskScheduler::~TaskScheduler() { + Shutdown(); + NEARBY_LOGS(INFO) << __func__ << ": Destroyed task scheduler: " << this; +} + +std::shared_ptr TaskScheduler::Schedule( + Runnable&& runnable, absl::Duration duration) { + return Schedule(std::move(runnable), duration, absl::ZeroDuration()); +} + +std::shared_ptr TaskScheduler::Schedule( + Runnable&& runnable, absl::Duration duration, + absl::Duration repeat_interval) { + MutexLock lock(&mutex_); + NEARBY_LOGS(INFO) << __func__ + << ": Scheduling task on task scheduler:" << this + << ", duration: " << absl::ToInt64Milliseconds(duration) + << "ms, repeat_interval: " + << absl::ToInt64Milliseconds(repeat_interval) << "ms"; + + std::shared_ptr task = + std::make_shared(*this, std::move(runnable)); + + HANDLE timer_handle = nullptr; + if (!CreateTimerQueueTimer(&timer_handle, nullptr, + static_cast(TimerRoutine), + task->runnable(), + absl::ToInt64Milliseconds(duration), + absl::ToInt64Milliseconds(repeat_interval), 0)) { + NEARBY_LOGS(ERROR) + << __func__ + << ": Failed to create timer queue timer in task scheduler:" << this + << " error: " << GetLastError(); + return nullptr; + } + + task->SetTimerHandle(reinterpret_cast(timer_handle)); + scheduled_tasks_.insert({reinterpret_cast(timer_handle), task}); + NEARBY_LOGS(INFO) << __func__ << ": Scheduled task " << task.get() + << " on task scheduler:" << this + << " timer handle: " << task->timer_handle(); + return task; +} + +void TaskScheduler::Shutdown() { + MutexLock lock(&mutex_); + NEARBY_LOGS(INFO) << __func__ << ": Shutting down task scheduler:" << this; + if (is_shutdown_) { + return; + } + for (auto& task : scheduled_tasks_) { + // Wait for running task to finish. + if (!DeleteTimerQueueTimer( + nullptr, reinterpret_cast(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_->remove_scheduled_task(timer_handle_); +} + +void TaskScheduler::ScheduledTask::SetTimerHandle(intptr_t timer_handle) { + timer_handle_ = timer_handle; +} + +Runnable* TaskScheduler::ScheduledTask::runnable() { return &runnable_; } + +intptr_t TaskScheduler::ScheduledTask::timer_handle() { return timer_handle_; } + +bool TaskScheduler::remove_scheduled_task(intptr_t timer_handle) { + MutexLock lock(&mutex_); + auto it = scheduled_tasks_.find(timer_handle); + if (it == scheduled_tasks_.end()) { + return false; + } + + // Wait for running task to finish. + if (!DeleteTimerQueueTimer(nullptr, reinterpret_cast(timer_handle), + INVALID_HANDLE_VALUE)) { + if (GetLastError() != ERROR_IO_PENDING) { + NEARBY_LOGS(ERROR) << __func__ << ": Failed to delete timer queue timer: " + << timer_handle << " error: " << GetLastError(); + } + } + + scheduled_tasks_.erase(it); + return true; +} + +} // namespace nearby::windows diff --git a/internal/platform/implementation/windows/task_scheduler.h b/internal/platform/implementation/windows/task_scheduler.h new file mode 100644 index 00000000..a3f4167e --- /dev/null +++ b/internal/platform/implementation/windows/task_scheduler.h @@ -0,0 +1,78 @@ +// Copyright 2024 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 THIRD_PARTY_NEARBY_INTERNAL_PLATFORM_IMPLEMENTATION_WINDOWS_TASK_SCHEDULER_H_ +#define THIRD_PARTY_NEARBY_INTERNAL_PLATFORM_IMPLEMENTATION_WINDOWS_TASK_SCHEDULER_H_ + +#include +#include + +#include "absl/base/thread_annotations.h" +#include "absl/container/flat_hash_map.h" +#include "absl/time/time.h" +#include "internal/platform/implementation/cancelable.h" +#include "internal/platform/mutex.h" +#include "internal/platform/runnable.h" + +namespace nearby::windows { + +// TaskScheduler is a utility class to scheduled a runnable task. It is used by +// the ScheduledExecutor and timer implementations. +class TaskScheduler { + public: + TaskScheduler(); + ~TaskScheduler(); + + std::shared_ptr Schedule(Runnable&& runnable, + absl::Duration duration); + + std::shared_ptr Schedule(Runnable&& runnable, + absl::Duration duration, + absl::Duration repeat_interval) + ABSL_LOCKS_EXCLUDED(mutex_); + + void Shutdown() ABSL_LOCKS_EXCLUDED(mutex_); + + private: + class ScheduledTask : public api::Cancelable { + public: + explicit ScheduledTask(TaskScheduler& task_scheduler, Runnable&& runnable); + ~ScheduledTask() override = default; + + // Note: not support to cancel and shutdown a scheduled task in the + // callback. + bool Cancel() override; + + void SetTimerHandle(intptr_t timer_handle); + intptr_t timer_handle(); + + Runnable* runnable(); + + private: + TaskScheduler* task_scheduler_; + Runnable runnable_; + intptr_t timer_handle_; + }; + + bool remove_scheduled_task(intptr_t timer_handle) ABSL_LOCKS_EXCLUDED(mutex_); + + Mutex mutex_; + bool is_shutdown_ ABSL_GUARDED_BY(mutex_) = false; + absl::flat_hash_map> scheduled_tasks_ + ABSL_GUARDED_BY(mutex_); +}; + +} // namespace nearby::windows + +#endif // THIRD_PARTY_NEARBY_INTERNAL_PLATFORM_IMPLEMENTATION_WINDOWS_TASK_SCHEDULER_H_ diff --git a/internal/platform/implementation/windows/task_scheduler_test.cc b/internal/platform/implementation/windows/task_scheduler_test.cc new file mode 100644 index 00000000..bc762fa2 --- /dev/null +++ b/internal/platform/implementation/windows/task_scheduler_test.cc @@ -0,0 +1,113 @@ +// Copyright 2024 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. + +#include "internal/platform/implementation/windows/task_scheduler.h" + +#include +#include + +#include "gtest/gtest.h" +#include "absl/synchronization/notification.h" +#include "absl/time/clock.h" +#include "absl/time/time.h" + +namespace nearby::windows { + +namespace { +constexpr absl::Duration kTaskDuration = absl::Milliseconds(500); + +TEST(TaskScheduler, ScheduleOneTask) { + TaskScheduler task_scheduler; + int counter = 0; + absl::Notification notification; + auto task = [&counter, ¬ification]() { + counter++; + notification.Notify(); + }; + auto cancelable = task_scheduler.Schedule(task, absl::Milliseconds(100)); + EXPECT_NE(cancelable, nullptr); + EXPECT_TRUE(notification.WaitForNotificationWithTimeout(kTaskDuration)); + EXPECT_EQ(counter, 1); +} + +TEST(TaskScheduler, ScheduleOneEarlierTaskAfterOneTask) { + TaskScheduler task_scheduler; + std::vector result; + absl::Notification notification1; + absl::Notification notification2; + auto task1 = [&result, ¬ification1]() { + result.push_back(1); + notification1.Notify(); + }; + + auto task2 = [&result, ¬ification2]() { + result.push_back(2); + notification2.Notify(); + }; + task_scheduler.Schedule(task1, absl::Milliseconds(100)); + task_scheduler.Schedule(task2, absl::Milliseconds(50)); + EXPECT_TRUE(notification1.WaitForNotificationWithTimeout(kTaskDuration)); + EXPECT_TRUE(notification2.WaitForNotificationWithTimeout(kTaskDuration)); + ASSERT_EQ(result.size(), 2); + EXPECT_EQ(result.at(0), 2); + EXPECT_EQ(result.at(1), 1); +} + +TEST(TaskScheduler, CancelScheduledTask) { + TaskScheduler task_scheduler; + int counter = 0; + absl::Notification notification; + auto task = [&counter, ¬ification]() { + counter++; + notification.Notify(); + }; + auto cancelable = task_scheduler.Schedule(task, absl::Milliseconds(200)); + cancelable->Cancel(); + EXPECT_FALSE(notification.WaitForNotificationWithTimeout(kTaskDuration)); + EXPECT_EQ(counter, 0); +} + +TEST(TaskScheduler, ShundownShouldWaitForScheduledTaskToFinish) { + TaskScheduler task_scheduler; + int counter = 0; + absl::Notification notification; + auto task = [&counter, ¬ification]() { + absl::SleepFor(kTaskDuration); + counter++; + notification.Notify(); + }; + auto cancelable = task_scheduler.Schedule(task, absl::Milliseconds(100)); + absl::SleepFor(absl::Milliseconds(200)); + task_scheduler.Shutdown(); + EXPECT_EQ(counter, 1); +} + +TEST(TaskScheduler, CancelCancleledScheduledTask) { + TaskScheduler task_scheduler; + int counter = 0; + absl::Notification notification; + auto task = [&counter, ¬ification]() { + absl::SleepFor(kTaskDuration); + counter++; + notification.Notify(); + }; + auto cancelable = task_scheduler.Schedule(task, absl::Milliseconds(200)); + EXPECT_TRUE(cancelable->Cancel()); + EXPECT_FALSE(cancelable->Cancel()); + task_scheduler.Shutdown(); + EXPECT_EQ(counter, 0); +} + +} // namespace +} // namespace nearby::windows